tests/test_analysis_single_source.py

from core.contracts import LayerPriority, Span
from core.factory import build_engine_from_recognizers
from core.output.trace import ClusterTrace
from web.services.analysis import AnalysisService, _llm_status, build_debug_payload
from web.services.cache import TTLCache
from web.services.documents import DocumentStore, StoredDocument


def _trace(text, kept):
    return ClusterTrace(
        text=text, entity_type="ORGANIZZAZIONE", label="Org", severity="media",
        start=0, end=len(text), base_score=0.9, agreement_boost=0.0,
        final_score=0.9 if kept else 0.0, agreement_count=1, validated=False,
        threshold=0.5, passed_threshold=kept, kept=kept,
    )


def test_stored_document_carries_traces_and_debug_metadata():
    sp = Span(0, 6, "Teatro", "ORGANIZZAZIONE", 0.9, LayerPriority.NER)
    doc = StoredDocument(
        doc_id="abc", text="Teatro", spans=[sp],
        traces=(_trace("Teatro", True), _trace("MUSICALI", False)),
        threshold=0.5, method="default", llm_status="active",
    )
    assert [t.text for t in doc.traces] == ["Teatro", "MUSICALI"]
    assert doc.threshold == 0.5
    assert doc.method == "default"
    assert doc.llm_status == "active"


def test_stored_document_debug_metadata_defaults():
    doc = StoredDocument(doc_id="abc", text="x", spans=[])
    assert doc.traces == ()
    assert doc.threshold == 0.0
    assert doc.method == ""
    assert doc.llm_status == "off"


def test_llm_status_three_states():
    assert _llm_status(verify=False, has_verifier=False) == "off"
    assert _llm_status(verify=True, has_verifier=False) == "unavailable"
    assert _llm_status(verify=True, has_verifier=True) == "active"


def test_build_debug_payload_counts_kept_and_rejected():
    kept = _trace("Teatro", True)
    rejected = ClusterTrace(
        text="MUSICALI", entity_type="ORGANIZZAZIONE", label="Org",
        severity="media", start=0, end=8, base_score=0.9, agreement_boost=0.0,
        final_score=0.0, agreement_count=1, validated=False, threshold=0.5,
        passed_threshold=False, kept=False, llm_vote=-1.0,
    )
    payload = build_debug_payload(
        (kept, rejected), threshold=0.5, method="default", llm_status="active"
    )
    assert payload["method"] == "default"
    assert payload["threshold"] == 0.5
    assert len(payload["clusters"]) == 2
    # discarded_reason serializzato a mano (non è un campo del dataclass)
    reasons = {c["text"]: c["discarded_reason"] for c in payload["clusters"]}
    assert reasons["MUSICALI"] == "sotto soglia di confidenza"
    s = payload["summary"]
    assert s["total_clusters"] == 2
    assert s["kept"] == 1
    assert s["below_threshold"] == 1
    assert s["llm_active"] is True
    assert s["llm_status"] == "active"
    assert s["llm_scored"] == 1
    assert s["llm_rejected"] == 1


class _Rec:
    """Riconoscitore fittizio: ritorna span fissi (Liskov)."""

    def __init__(self, spans, layer=LayerPriority.NER, name="fake"):
        self._spans = spans
        self.layer = layer
        self.name = name

    def analyze(self, text):  # noqa: ARG002
        return self._spans


class _CountingVerifier:
    """Giudice simulato: boccia un testo noto (score 0) e conta le chiamate."""

    def __init__(self, reject_text):
        self.calls = 0
        self._reject = reject_text

    def verify_with_votes(self, text, spans, on_progress=None):  # noqa: ARG002
        self.calls += 1
        out, votes, prompts = [], {}, {}
        for i, sp in enumerate(spans):
            prompts[i] = f"prompt::{sp.text}"
            if sp.text == self._reject:
                out.append(sp.with_changes(score=0.0))
                votes[i] = -1.0
            else:
                out.append(sp)
                votes[i] = 1.0
        return out, votes, prompts

    def verify(self, text, spans):
        return self.verify_with_votes(text, spans)[0]


class _StubRegistry:
    """Registry minimale per AnalysisService: un solo engine, store reale."""

    def __init__(self, engine):
        self._engine = engine
        self.documents = DocumentStore(TTLCache())
        self.default_method = "default"

    def engine_for(self, method=None, verify=False, approver=None,
                   simulate=False):  # noqa: ARG002
        return self._engine

    def engine_for_layers(self, layers, verify=False, approver=None,
                          simulate=False):  # noqa: ARG002
        return self._engine

    def process_layer_index(self):
        # Nessuna famiglia cache-ata nello stub: i tempi per-processo finiscono
        # come voci "orfane" (col nome del processo) — vedi _aggregate_timings.
        return {}

    @property
    def engine(self):
        return self._engine


def _service_with_judge():
    text = "Teatro strumenti MUSICALI"
    spans = [
        Span(0, 6, "Teatro", "ORGANIZZAZIONE", 0.9, LayerPriority.NER),
        Span(17, 25, "MUSICALI", "ORGANIZZAZIONE", 0.9, LayerPriority.NER),
    ]
    verifier = _CountingVerifier("MUSICALI")
    engine = build_engine_from_recognizers([_Rec(spans)], verifier=verifier)
    return AnalysisService(_StubRegistry(engine)), text, verifier


def test_analyze_text_stores_traces_and_excludes_rejected():
    service, text, verifier = _service_with_judge()
    doc_id, render = service.analyze_text(text, min_confidence=0.5, verify=True)
    stored = service._registry.documents.get(doc_id)
    # La passata col giudice ha salvato le tracce (incluso il cluster scartato).
    assert {t.text for t in stored.traces} == {"Teatro", "MUSICALI"}
    # Gli span oscurati NON contengono la parola bocciata dal giudice.
    assert {s.text for s in stored.spans} == {"Teatro"}
    assert render.report.total == 1
    # Invariante: il giudice è stato eseguito UNA sola volta (passata unica).
    assert verifier.calls == 1


def test_min_confidence_in_salt_separates_cache_entries():
    service, text, _ = _service_with_judge()
    id_a, _ = service.analyze_text(text, min_confidence=0.5, verify=True)
    id_b, _ = service.analyze_text(text, min_confidence=0.95, verify=True)
    assert id_a != id_b


def test_debug_reuses_analyze_pass_judge_called_once():
    service, text, verifier = _service_with_judge()
    doc_id, _ = service.analyze_text(text, min_confidence=0.5, verify=True)
    payload = service.debug_trace(text, min_confidence=0.5, verify=True)
    # Il giudice NON è stato rieseguito dal debug: una sola passata totale.
    assert verifier.calls == 1
    # Il debug vede esattamente lo stesso esito della cache.
    reasons = {c["text"]: c["kept"] for c in payload["clusters"]}
    assert reasons == {"Teatro": True, "MUSICALI": False}
    assert payload["summary"]["llm_rejected"] == 1


def test_debug_for_doc_matches_rendered_spans():
    service, text, _ = _service_with_judge()
    doc_id, render = service.analyze_text(text, min_confidence=0.5, verify=True)
    payload = service.debug_for_doc(doc_id)
    kept_texts = {c["text"] for c in payload["clusters"] if c["kept"]}
    stored = service._registry.documents.get(doc_id)
    oscurati = {s.text for s in stored.spans}
    # INVARIANTE: cluster kept == span oscurati.
    assert kept_texts == oscurati == {"Teatro"}


def test_debug_for_doc_missing_returns_none():
    service, _text, _ = _service_with_judge()
    assert service.debug_for_doc("inesistente") is None


def test_request_key_layers_and_default():
    service, _text, _ = _service_with_judge()
    assert service.request_key(None, None) == "default"
    assert service.request_key("rules", None) == "rules"
    assert service.request_key(None, ["ner", "rules"]) == "layers:ner,rules"
    assert service.request_key(None, []) == "layers:rules"


def test_pdf_flow_single_pass_invariant():
    """Flusso PDF: una sola passata del giudice e debug==oscurati (scenario del bug)."""
    service, text, verifier = _service_with_judge()
    doc_id = "pdfdoc"
    service.store_pdf_detection(
        doc_id, text, method=None, layers=None, verify=True, approver=None,
        min_confidence=0.5, pages=(), is_scanned=False, raw_bytes=b"%PDF-1.4",
    )
    assert verifier.calls == 1  # giudice eseguito una sola volta
    stored = service._registry.documents.get(doc_id)
    assert stored.is_pdf is True
    assert stored.raw_bytes == b"%PDF-1.4"
    payload = service.debug_for_doc(doc_id)
    kept_texts = {c["text"] for c in payload["clusters"] if c["kept"]}
    # INVARIANTE: cluster kept del debug == span oscurati salvati.
    assert kept_texts == {s.text for s in stored.spans} == {"Teatro"}
    assert payload["summary"]["llm_rejected"] == 1


def test_llm_check_for_reports_active_status_and_counts():
    service, text, _ = _service_with_judge()
    doc_id, _ = service.analyze_text(text, min_confidence=0.5, verify=True)
    check = service.llm_check_for(doc_id)
    assert check["status"] == "active"
    assert check["total"] == 2           # Teatro + MUSICALI
    assert check["scored"] == 2          # entrambe hanno un voto
    assert check["obscured_without_score"] == 0  # nessuna tenuta senza voto


def test_llm_check_for_missing_doc_returns_none():
    service, _text, _ = _service_with_judge()
    assert service.llm_check_for("inesistente") is None


def _service_no_detections():
    verifier = _CountingVerifier("qualsiasi")
    engine = build_engine_from_recognizers([_Rec([])], verifier=verifier)
    return AnalysisService(_StubRegistry(engine)), "testo senza entita", verifier


def test_analyzed_empty_doc_debug_is_empty_not_none_and_not_rerun():
    """Un doc analizzato SENZA rilevamenti: debug valido-ma-vuoto (non None) e
    nessun rilancio della detection (traces=() ⇒ 'analizzato', non 'da analizzare')."""
    service, text, verifier = _service_no_detections()
    doc_id, render = service.analyze_text(text, min_confidence=0.5, verify=True)
    assert render.report.total == 0
    payload = service.debug_for_doc(doc_id)
    assert payload is not None
    assert payload["summary"]["total_clusters"] == 0
    calls_before = verifier.calls
    service.debug_trace(text, min_confidence=0.5, verify=True)
    assert verifier.calls == calls_before  # nessuna riesecuzione


def test_aggregate_timings_groups_by_layer_with_judge_last():
    from web.services.analysis import _aggregate_timings

    sink = {
        "l0_cf": 0.2, "l0_iban": 0.1,          # due processi del layer regole
        "ner": 1.0, "reclass(ner)": 0.5,        # due processi dello stesso layer NER
        "processo_ignoto": 0.3,                 # famiglia non in cache (mappa incompleta)
        "__giudice__": 3.0,
    }
    idx = {
        "l0_cf": ("rules", "Regole"), "l0_iban": ("rules", "Regole"),
        "ner": ("ner", "NER GDPR"), "reclass(ner)": ("ner", "NER GDPR"),
    }
    out = _aggregate_timings(sink, idx)

    by_key = {e["key"]: e for e in out}
    assert by_key["rules"]["seconds"] == 0.3          # somma dei processi del layer
    assert by_key["ner"]["seconds"] == 1.5
    # L'etichetta è quella canonica del catalogo layer (non quella della mappa).
    from config.detection.layers import DETECTION_LAYERS
    catalog_labels = {s.key: s.label for s in DETECTION_LAYERS}
    assert by_key["ner"]["label"] == catalog_labels["ner"]
    # Il processo orfano non va perso: compare col proprio nome.
    assert by_key["processo_ignoto"]["seconds"] == 0.3
    # Il giudice è SEMPRE l'ultima voce.
    assert out[-1] == {"key": "giudice", "label": "Giudice LLM", "seconds": 3.0}
    # I layer noti rispettano l'ordine del catalogo (rules prima di ner).
    keys = [e["key"] for e in out]
    assert keys.index("rules") < keys.index("ner")


def test_detect_and_store_saves_timings():
    service, text, _verifier = _service_with_judge()
    doc_id, _ = service.analyze_text(text, verify=True)
    stored = service._registry.documents.get(doc_id)
    # Lo stub non ha famiglie in cache: i tempi compaiono per-processo, e con
    # il giudice agganciato c'è SEMPRE la voce finale "giudice".
    assert stored.timings, "timings assenti sul documento"
    assert stored.timings[-1]["key"] == "giudice"
    assert all(
        set(t) == {"key", "label", "seconds"} and t["seconds"] >= 0
        for t in stored.timings
    )