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

"""Regression test for #136 — an unlinked digest citation must never be dropped
silently.

``_handle_unlinked_citation`` routes via the canonical classifier:
  * supreme/admin → a court-fetch job (no missing_precedent here — the X13
    orchestrator opens its own on failure),
  * skip (ערר/בל"מ) / unknown → a deduped missing_precedent (discovery_source
    'digest'), which previously vanished.

Runs OFFLINE — monkeypatches the db calls and records what each routing did.
"""

from __future__ import annotations

import asyncio

import pytest

from legal_mcp.services import digest_library as dl
from legal_mcp.services import db


class _Spy:
    def __init__(self):
        self.court_fetch = []
        self.created_mp = []
        self.find_mp_returns = None

    def install(self, monkeypatch):
        async def _job_upsert(**kw):
            self.court_fetch.append(kw)
        async def _find_mp(citation, case_id=None):
            return self.find_mp_returns
        async def _create_mp(**kw):
            self.created_mp.append(kw)
            return {"id": "mp"}
        async def _get_digest(_id):
            return {"yomon_number": "5167"}
        monkeypatch.setattr(db, "court_fetch_job_upsert", _job_upsert)
        monkeypatch.setattr(db, "find_missing_precedent_by_citation", _find_mp)
        monkeypatch.setattr(db, "create_missing_precedent", _create_mp)
        monkeypatch.setattr(db, "get_digest", _get_digest)


def _run(coro):
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()


@pytest.fixture()
def spy(monkeypatch):
    s = _Spy()
    s.install(monkeypatch)
    return s


_DID = "11111111-1111-1111-1111-111111111111"


def test_committee_citation_opens_missing_precedent(spy):
    _run(dl._handle_unlinked_citation(_DID, "ערר 1198-12-25 זאטוס"))
    assert spy.court_fetch == []                       # ערר is never auto-fetched
    assert len(spy.created_mp) == 1, spy.created_mp
    mp = spy.created_mp[0]
    assert mp["discovery_source"] == "digest"
    assert "יומון" in (mp["notes"] or "")             # provenance recorded


def test_court_verdict_enqueues_fetch_not_mp(spy):
    _run(dl._handle_unlinked_citation(_DID, 'עע"מ 3975/22 פלוני'))
    assert len(spy.court_fetch) == 1, spy.court_fetch
    assert spy.created_mp == []                        # fetchable → orchestrator owns its MP


def test_dedup_skips_existing_gap(spy):
    spy.find_mp_returns = {"id": "existing"}           # gap already recorded
    _run(dl._handle_unlinked_citation(_DID, "ערר 1192/18"))
    assert spy.created_mp == []                        # no duplicate


def test_unknown_citation_opens_missing_precedent(spy):
    _run(dl._handle_unlinked_citation(_DID, "משהו בלי ערכאה ברורה"))
    # unknown tier is not fetchable → must still surface as a gap, never dropped.
    assert spy.court_fetch == []
    assert len(spy.created_mp) == 1