pdf_oralia_vibe/tests/test_parsers_golden.py

"""Tests de non-regression des parseurs sur les PDF reels.

Les parseurs sont le coeur du produit et la partie la plus exposee aux
regressions silencieuses : un decalage de colonne ne leve aucune exception, il
produit juste des montants faux. Ces tests figent, pour chaque PDF du corpus
local, une **empreinte** de l'extraction (structure des lots, totaux par lot,
montants par categorie d'operation) et signalent tout ecart.

Les comptes rendus contiennent des donnees personnelles : ni les PDF (`data/`)
ni les references generees (`tests/golden/`) ne sont versionnes. La suite se
saute donc d'elle-meme la ou le corpus est absent, CI comprise. Les libelles
sensibles (locataire, fournisseur, description) sont reduits a une empreinte
courte : un changement reste detecte, sans recopier la donnee.

Regenerer les references apres un changement volontaire de parseur :

    uv run pytest tests/test_parsers_golden.py --regen-golden

Puis relire le `git diff`... qui n'existe pas ici : comparer a la main la sortie
avant/apres, ou versionner temporairement le dossier pour l'inspecter.
"""

import hashlib
import json
from pathlib import Path

import pytest

from plesna_gerance.parsers.locataires_table import (
    extract_situation_locataires_from_pdf,
)
from plesna_gerance.parsers.operations_table import (
    extract_recapitulatif_operations_from_pdf,
)

_RACINE = Path(__file__).resolve().parent.parent
_CORPUS = _RACINE / "data" / "documents"
_GOLDEN = Path(__file__).resolve().parent / "golden"

_MONTANTS_OPERATION = ("debit", "credit", "tva", "locatif", "deductible")


def _pdfs() -> list[Path]:
    """PDF du corpus local, tries pour un ordre de test stable."""
    if not _CORPUS.is_dir():
        return []
    return sorted(_CORPUS.rglob("*.pdf"))


def _empreinte_libelle(valeur: str | None) -> str | None:
    """Empreinte courte d'un libelle sensible (nom, fournisseur, description).

    Detecte toute modification du libelle sans en conserver le contenu.
    """
    if not valeur:
        return None
    normalise = " ".join(valeur.split())
    return hashlib.sha1(normalise.encode("utf-8")).hexdigest()[:8]


def _empreinte_lot(situation: dict) -> dict:
    types_lignes: dict[str, int] = {}
    for ligne in situation.get("lignes", []):
        type_ligne = ligne.get("type", "?")
        types_lignes[type_ligne] = types_lignes.get(type_ligne, 0) + 1

    return {
        "numero": situation.get("lot", {}).get("numero"),
        "type": situation.get("lot", {}).get("type"),
        "locataire": _empreinte_libelle(situation.get("locataire", {}).get("nom")),
        "nb_lignes": len(situation.get("lignes", [])),
        "types_lignes": dict(sorted(types_lignes.items())),
        "totaux": situation.get("totaux", {}),
    }


def _empreinte_operation(operation: dict) -> dict:
    return {
        "categorie": operation.get("categorie"),
        "sous_categorie": _empreinte_libelle(operation.get("sous_categorie")),
        "fournisseur": _empreinte_libelle(operation.get("fournisseur")),
        "description": _empreinte_libelle(operation.get("description")),
        "lot_numero": operation.get("lot_numero"),
        "montants": operation.get("montants", {}),
    }


def _empreinte(situations: list[dict], operations: list[dict]) -> dict:
    """Empreinte complete d'une extraction, comparable d'une execution a l'autre."""
    par_categorie: dict[str, dict] = {}
    for operation in operations:
        categorie = operation.get("categorie") or "SANS_CATEGORIE"
        agrege = par_categorie.setdefault(
            categorie, {"nb": 0, **{cle: 0.0 for cle in _MONTANTS_OPERATION}}
        )
        agrege["nb"] += 1
        for cle in _MONTANTS_OPERATION:
            agrege[cle] = round(
                agrege[cle] + (operation.get("montants", {}).get(cle) or 0.0), 2
            )

    return {
        "situation_locataires": {
            "nb_lots": len(situations),
            "lots": [_empreinte_lot(situation) for situation in situations],
        },
        "recapitulatif_operations": {
            "nb_operations": len(operations),
            "par_categorie": dict(sorted(par_categorie.items())),
            "operations": [_empreinte_operation(op) for op in operations],
        },
    }


@pytest.mark.skipif(not _pdfs(), reason="corpus PDF local absent (data/documents)")
@pytest.mark.parametrize("pdf", _pdfs(), ids=lambda p: p.stem)
def test_extraction_conforme_a_la_reference(pdf: Path, regen_golden: bool):
    """L'extraction d'un PDF reel reste identique a sa reference figee."""
    obtenue = _empreinte(
        extract_situation_locataires_from_pdf(str(pdf)),
        extract_recapitulatif_operations_from_pdf(str(pdf)),
    )

    reference = _GOLDEN / f"{pdf.stem}.json"

    if regen_golden:
        _GOLDEN.mkdir(parents=True, exist_ok=True)
        reference.write_text(
            json.dumps(obtenue, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
        )
        pytest.skip(f"reference regeneree : {reference.name}")

    if not reference.exists():
        pytest.fail(
            f"Reference absente pour {pdf.name}. "
            "Generer avec : uv run pytest tests/test_parsers_golden.py --regen-golden"
        )

    assert obtenue == json.loads(reference.read_text(encoding="utf-8"))


@pytest.mark.skipif(not _pdfs(), reason="corpus PDF local absent (data/documents)")
def test_le_parseur_geometrique_couvre_tout_le_corpus():
    """Aucun PDF du corpus ne ressort vide du parseur geometrique.

    Verrouille la decision de supprimer les parseurs texte de repli : le jour ou
    un PDF echappe au parseur par cellules, ce test le signale plutot que de
    laisser passer un document vide.
    """
    vides = [
        pdf.name
        for pdf in _pdfs()
        if not extract_situation_locataires_from_pdf(str(pdf))
    ]
    assert vides == []