legal-ai/mcp-server/tests/test_party_claims_summary.py

"""Tests for the party-claims executive summary (#202, WS3).

Cover the deterministic, offline parts:
  - ``_build_arguments_context`` source selection: legal_arguments preferred,
    raw claims fallback, "none" when empty (G2 — single source, no re-extract).
  - ``summarize_party_claims`` pins Opus 4.8 + effort=high and grounds the
    prompt strictly in the case's own claims (anti-hallucination / INV-AH).
  - ``_render_markdown_body`` maps markdown headings/bullets onto template styles.

Generation itself (claude -p) is local-only and not exercised here — the
claude_session call is monkeypatched.
"""

from __future__ import annotations

import asyncio
from uuid import uuid4

from legal_mcp.services import party_claims_summary as pcs


def _run(coro):
    return asyncio.run(coro)


# ── source selection (G2: arguments preferred, claims fallback) ────


def test_arguments_preferred_over_claims(monkeypatch):
    cid = uuid4()

    async def _get_args(_case_id):
        return [
            {"party": "appellant", "argument_title": "חריגת בנייה",
             "argument_body": "הבנייה חורגת מקו הבניין", "legal_topic": "קווי בניין"},
            {"party": "respondent", "argument_title": "התיישנות",
             "argument_body": "", "legal_topic": ""},
        ]

    async def _get_claims(_case_id, *a, **k):  # must NOT be consulted
        raise AssertionError("claims fallback used despite arguments existing")

    monkeypatch.setattr(pcs.argument_aggregator, "get_legal_arguments", _get_args)
    monkeypatch.setattr(pcs.db, "get_claims", _get_claims)

    ctx, n, kind = _run(pcs._build_arguments_context(cid))
    assert kind == "arguments"
    assert n == 2
    assert "העוררים" in ctx and "המשיבים" in ctx
    assert "חריגת בנייה" in ctx


def test_falls_back_to_claims_when_no_arguments(monkeypatch):
    cid = uuid4()

    async def _no_args(_case_id):
        return []

    async def _claims(_case_id, *a, **k):
        return [
            {"party_role": "appellant", "claim_text": "טענה אחת", "source_document": "appeal"},
            {"party_role": "respondent", "claim_text": "טענה שתיים", "source_document": "response"},
        ]

    monkeypatch.setattr(pcs.argument_aggregator, "get_legal_arguments", _no_args)
    monkeypatch.setattr(pcs.db, "get_claims", _claims)

    ctx, n, kind = _run(pcs._build_arguments_context(cid))
    assert kind == "claims"
    assert n == 2
    assert "טענה אחת" in ctx


def test_block_zayin_claims_excluded(monkeypatch):
    """Decision-summary (block-zayin) claims are not original pleadings."""
    cid = uuid4()

    async def _no_args(_case_id):
        return []

    async def _claims(_case_id, *a, **k):
        return [
            {"party_role": "appellant", "claim_text": "מקורית", "source_document": "appeal"},
            {"party_role": "appellant", "claim_text": "מסיכום", "source_document": "block-zayin"},
        ]

    monkeypatch.setattr(pcs.argument_aggregator, "get_legal_arguments", _no_args)
    monkeypatch.setattr(pcs.db, "get_claims", _claims)

    ctx, n, kind = _run(pcs._build_arguments_context(cid))
    assert kind == "claims"
    assert n == 1
    assert "מקורית" in ctx and "מסיכום" not in ctx


def test_none_when_no_source(monkeypatch):
    cid = uuid4()
    monkeypatch.setattr(pcs.argument_aggregator, "get_legal_arguments",
                        lambda _c: _aw([]))
    monkeypatch.setattr(pcs.db, "get_claims", lambda _c, *a, **k: _aw([]))
    ctx, n, kind = _run(pcs._build_arguments_context(cid))
    assert kind == "none" and n == 0 and ctx == ""


async def _aw(v):
    return v


# ── generation pins Opus 4.8 + high effort, grounded in claims ─────


def test_summarize_pins_opus_and_high_effort(monkeypatch, tmp_path):
    cid = uuid4()
    captured: dict = {}

    async def _get_case(_case_id):
        return {"case_number": "8125-09-24", "title": "t", "appellants": ["א"],
                "respondents": ["ב"], "subject": "היטל השבחה", "property_address": "",
                "appeal_type": "betterment_levy", "practice_area": "betterment_levy"}

    async def _get_decision(_case_id):
        return None

    async def _args(_case_id):
        return [{"party": "appellant", "argument_title": "X",
                 "argument_body": "Y", "legal_topic": ""}]

    async def _fake_query(prompt, *, timeout=None, model=None, effort=None, tools=None):
        captured["model"] = model
        captured["effort"] = effort
        captured["tools"] = tools
        captured["prompt"] = prompt
        return "## טענות העוררים\nתמצית."

    async def _style(_practice_area=""):
        return "(style)"

    monkeypatch.setattr(pcs.db, "get_case", _get_case)
    monkeypatch.setattr(pcs.db, "get_decision_by_case", _get_decision)
    monkeypatch.setattr(pcs.argument_aggregator, "get_legal_arguments", _args)
    monkeypatch.setattr(pcs.block_writer, "_build_style_context", _style)
    monkeypatch.setattr(pcs.claude_session, "query", _fake_query)
    monkeypatch.setattr(pcs.config, "find_case_dir", lambda cn: tmp_path / cn)
    monkeypatch.setattr(pcs.config, "DATA_DIR", tmp_path)

    async def _mirror(*a, **k):
        return None
    monkeypatch.setattr(pcs.storage, "mirror", _mirror)
    monkeypatch.setattr(pcs.git_sync, "commit_and_push", lambda *a, **k: True)

    result = _run(pcs.summarize_party_claims(cid, "8125-09-24"))

    # The whole point of #202's generation-path constraint:
    assert captured["model"] == "claude-opus-4-8"
    assert captured["effort"] == "high"
    assert captured["tools"] == ""  # prose, no tool_use
    # Grounded strictly in the case's own arguments (anti-hallucination).
    assert "אנטי-הזיה" in captured["prompt"]
    assert "Y" in captured["prompt"]  # the actual argument body is in-context

    assert result["source_kind"] == "arguments"
    out = (tmp_path / "8125-09-24" / "documents" / "research" / "party-claims-summary.md")
    assert out.exists()
    assert "סיכום-מנהלים" in out.read_text(encoding="utf-8")


def test_summarize_raises_when_no_claims(monkeypatch, tmp_path):
    cid = uuid4()
    monkeypatch.setattr(pcs.db, "get_case", lambda _c: _aw({"case_number": "9000-01-25"}))
    monkeypatch.setattr(pcs.db, "get_decision_by_case", lambda _c: _aw(None))
    monkeypatch.setattr(pcs.argument_aggregator, "get_legal_arguments", lambda _c: _aw([]))
    monkeypatch.setattr(pcs.db, "get_claims", lambda _c, *a, **k: _aw([]))

    try:
        _run(pcs.summarize_party_claims(cid, "9000-01-25"))
        raise AssertionError("expected ValueError on empty source")
    except ValueError as e:
        assert "טענות" in str(e)


# ── markdown → DOCX rendering uses template heading styles ─────────


def test_render_markdown_body_maps_styles():
    from docx import Document
    from legal_mcp.services import analysis_docx_exporter as ax

    doc = Document()
    md = (
        "# כותרת ראשית\n\n"
        "---\n"
        "## טענות העוררים\n"
        "- טענה אחת\n"
        "1. סעיף ממוספר\n"
        "טקסט רגיל.\n"
    )
    ax._render_markdown_body(doc, md)
    styles = [p.style.name for p in doc.paragraphs]
    texts = [p.text for p in doc.paragraphs]
    # HR is skipped; headings become Heading styles.
    assert "כותרת ראשית" in texts
    assert "Heading 1" in styles  # # / ## → Heading 1
    assert any("טענה אחת" in t for t in texts)
    assert not any(t.strip() in {"---", "#"} for t in texts)


def test_next_version_is_prefix_scoped(tmp_path):
    from legal_mcp.services import analysis_docx_exporter as ax
    (tmp_path / "ניתוח-משפטי-v3.docx").write_bytes(b"x")
    # An analysis file at v3 must NOT bump the summary's version.
    assert ax._next_version(tmp_path, "סיכום-מנהלים-טענות") == 1
    assert ax._next_version(tmp_path, "ניתוח-משפטי") == 4