Spaces:

rahim237cm
/

jurix-api

Runtime error

App Files Files Community
jurix-api / tests /test_tasks /test_process_law.py
rahim237cm
Claude Opus 5.5
feat(embeddings): voyage-4-large remplace EmbeddingGemma (API Voyage, halfvec 1024)
57c8117 about 1 month ago
Raw History Blame Contribute Delete
24.2 kB
"""
Tests du pipeline de traitement des documents.
app/tasks/process_law.py fait ~700 lignes et n'avait aucun test : le repertoire
tests/test_tasks/ ne contenait qu'un __init__.py vide. C'est pourtant le seul
chemin par lequel un PDF devient des articles indexables.
Les tests marques integration touchent la base de test ; les autres sont de
pures fonctions de texte.
"""


import numpy as np
import pytest
from sqlalchemy import select
from sqlalchemy.orm import undefer

from app.models.law import Article, Law
from app.services.embedding_service import EmbeddingService
from app.services.legal_domain_classifier import (
    AFFAIRES,
    CANONICAL_DOMAINS,
    FINANCES,
    FONCTION_PUBLIQUE,
)
from app.tasks import process_law as pl


class _FakeSessionFactory:
    """
    Remplace SyncSessionLocal par une table `categories` en memoire.
    `_classify_category` n'a besoin que de la correspondance nom -> identifiant :
    la faire passer par une vraie base rendrait ces tests dependants de l'ordre
    des migrations, alors que ce qu'ils verifient est justement que le code ne
    depend PAS d'un ordre.
    """

    def __init__(self, rows):
        self._rows = rows

    def __call__(self):
        return self

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def execute(self, *args, **kwargs):
        rows = [(name, identifier) for name, identifier in self._rows.items()]

        class _Result:
            def all(self_inner):
                return rows

            def scalars(self_inner):
                class _Scalars:
                    def all(self_deep):
                        return [name for name, _ in rows]

                return _Scalars()

        return _Result()


# ==================== FONCTIONS PURES ====================


class TestTitleExtraction:

    @pytest.mark.parametrize("text,expected_start", [
        ("LOI N° 2024-001 DU 15 JANVIER 2024\nPortant régime des sociétés", "LOI N° 2024-001"),
        ("LAW N° 2024-002 OF 15 JANUARY 2024\nOn companies", "LAW N° 2024-002"),
        ("LA CONSTITUTION DE LA RÉPUBLIQUE DU CAMEROUN\nPréambule", "LA CONSTITUTION"),
    ])
    def test_extracts_known_patterns(self, text, expected_start):
        title = pl._extract_title_from_text(text)

        assert title is not None
        assert title.startswith(expected_start)

    def test_returns_none_on_empty_text(self):
        assert pl._extract_title_from_text("") is None

    def test_returns_none_on_junk(self):
        assert pl._extract_title_from_text("aaa\nbbb\nccc") is None


class TestTextCleaning:

    def test_rejoins_hyphenated_words(self):
        cleaned = pl._clean_extracted_text("respon-\nsabilité des dirigeants")

        assert "responsabilité" in cleaned

    def test_rejoins_elisions(self):
        cleaned = pl._clean_extracted_text("l'\narticle premier")

        assert "l'article" in cleaned

    def test_removes_page_markers(self):
        cleaned = pl._clean_extracted_text("Article 1.\n<<PAGE: 12>>\nContenu.")

        assert "<<PAGE" not in cleaned
        assert "Article 1." in cleaned
        assert "Contenu." in cleaned

    def test_removes_standalone_page_numbers(self):
        cleaned = pl._clean_extracted_text("Article 1.\n42\nContenu de l'article.")

        assert "\n42\n" not in cleaned

    def test_collapses_blank_lines(self):
        cleaned = pl._clean_extracted_text("A.\n\n\n\n\nB.")

        assert "\n\n\n" not in cleaned

    def test_empty_input(self):
        assert pl._clean_extracted_text("") == ""


class TestFallbacks:

    def test_language_detection_falls_back_on_error(self, monkeypatch):
        """Le pipeline ne doit pas s'arreter parce que le detecteur echoue."""
        import app.services.language_detector as detector_module

        class _Broken:
            def __init__(self, *a, **k):
                raise RuntimeError("modele absent")

        monkeypatch.setattr(detector_module, "LanguageDetector", _Broken)

        result = pl._detect_language("Le présent article de la loi et les décrets")

        assert result["language"] == "fr"
        assert 0.0 < result["confidence"] <= 1.0

    def test_category_classification_always_returns_a_domain(self, monkeypatch, classeur):
        """Le verdict du classifieur remonte tel quel, avec sa confiance."""
        classeur(AFFAIRES)
        monkeypatch.setattr(
            pl, "SyncSessionLocal", _FakeSessionFactory({}), raising=True
        )

        result = pl._classify_category("Texte quelconque", "Contenu quelconque")

        assert result["category"] == AFFAIRES
        assert result["confidence"] == 0.9


# ==================== INTEGRATION (base de test) ====================


@pytest.fixture
async def law_row(db_session):
    law = Law(
        reference="LOI-PIPE-001",
        title="Loi de test pipeline",
        content="Contenu initial.",
        type="loi",
        language="fr",
        status="processing",
    )
    db_session.add(law)
    await db_session.commit()
    await db_session.refresh(law)
    return law


# Empreinte du fournisseur double : distincte de toute empreinte reelle, pour
# qu'une ecriture qui irait la chercher ailleurs que dans le service se voie.
EMPREINTE_DOUBLURE = "doublure|test@0"


# Articles de longueur REALISTE : le raffinage ecarte du vectoriel tout chunk
# de moins de 120 caracteres, et c'est justifie — sur le corpus reel, 62 % des
# chunks sous ce seuil sont des lignes de liste nominative ("100. DOURLAI
# STANISLAS 0769502K"), des fragments de tableau ou des separateurs. Un
# echantillon de test fait d'articles de 70 caracteres mesurerait donc le seuil,
# pas le pipeline.
SAMPLE_TEXT = """
LOI N° 2024-500 DU 1 JUIN 2024
PORTANT DISPOSITIONS DIVERSES
CHAPITRE I - DISPOSITIONS GÉNÉRALES
Article 1. Objet
La présente loi fixe les règles applicables à la constitution, au fonctionnement
et à la dissolution des sociétés commerciales exerçant sur le territoire de la
République du Cameroun, sans préjudice des dispositions de l'Acte uniforme OHADA.
Article 2. Champ d'application
Elle s'applique sur l'ensemble du territoire national et concerne toutes les
sociétés commerciales immatriculées au registre du commerce et du crédit
mobilier, quelle que soit la nationalité de leurs associés ou de leurs dirigeants.
Article 3. Définitions
Au sens de la présente loi, on entend par société commerciale toute personne
morale constituée par deux ou plusieurs personnes qui conviennent d'affecter des
biens à une activité en vue de partager le bénéfice qui pourra en résulter.
"""


class TestArticlePersistence:

    @pytest.mark.asyncio
    async def test_saves_section_and_page_number(self, db_session, law_row):
        count = pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        assert count >= 3
        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id).order_by(Article.order)
        )).scalars().all()

        numbered = [a for a in rows if a.number.isdigit()]
        assert numbered
        assert any(a.section and "CHAPITRE" in a.section for a in numbered)
        assert all(a.content for a in numbered)

    @pytest.mark.asyncio
    async def test_embed_text_porte_la_categorie_des_la_premiere_ingestion(
        self, db_session, law_row
    ):
        """
        Le domaine est calcule avant le decoupage mais n'est ecrit en base
        qu'apres : relu en base, il manquait a embed_text a la premiere
        ingestion. Il est desormais transmis au decoupage.
        """
        pl._split_and_save_articles(
            law_row.id, SAMPLE_TEXT, language="fr", category=AFFAIRES
        )

        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id, Article.embed.is_(True))
        )).scalars().all()

        assert rows
        assert all(AFFAIRES in a.embed_text for a in rows)

    @pytest.mark.asyncio
    async def test_reprocessing_replaces_instead_of_duplicating(self, db_session, law_row):
        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)
        first = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()

        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)
        second = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()

        assert len(second) == len(first)


class TestEmbeddingGeneration:

    @pytest.fixture(autouse=True)
    def _stub_service(self, monkeypatch):
        """Doublure d'EmbeddingService : aucun appel reseau."""
        dim = EmbeddingService.EMBEDDING_DIM
        calls = {"texts": None, "fermetures": 0}

        class _Provider:
            label = "doublure"
            fingerprint = EMPREINTE_DOUBLURE

        class _Stub:
            MAX_TEXT_LENGTH = EmbeddingService.MAX_TEXT_LENGTH
            EMBEDDING_DIM = dim
            provider = _Provider()

            def __init__(self, *a, **k):
                pass

            def fermer(self):
                calls["fermetures"] += 1

            def generate_batch_embeddings(self, texts, **kwargs):
                calls["texts"] = texts
                rng = np.random.default_rng(1)
                out = []
                for _ in texts:
                    vec = rng.normal(size=dim)
                    out.append((vec / np.linalg.norm(vec)).astype(np.float32))
                return out

        import app.services.embedding_service as module

        monkeypatch.setattr(module, "EmbeddingService", _Stub)
        return calls

    @pytest.mark.asyncio
    async def test_writes_vectors_for_every_article(self, db_session, law_row, _stub_service):
        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        written = pl._generate_article_embeddings(law_row.id)

        assert written > 0
        # Un service par loi : sa connexion gardee ne survit pas a la loi.
        assert _stub_service["fermetures"] == 1
        # undefer : la colonne est differee, et une lecture paresseuse est
        # impossible en asynchrone (MissingGreenlet).
        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
            .options(undefer(Article.embedding))
        )).scalars().all()

        # Seuls les chunks marques embed sont vectorises : les visas et les
        # fragments restent en base, cherchables en plein texte, sans vecteur.
        embeddable = [a for a in rows if a.embed]
        skipped = [a for a in rows if not a.embed]

        assert embeddable, "au moins un chunk doit etre vectorisable"
        assert all(a.embedding is not None for a in embeddable)
        assert all(a.embedding is None for a in skipped)
        assert len(embeddable[0].embedding) == EmbeddingService.EMBEDDING_DIM

    @pytest.mark.asyncio
    async def test_each_vector_carries_its_provenance(self, db_session, law_row):
        """
        L'empreinte du fournisseur part avec chaque vecteur. Sans elle, un
        corpus a moitie regenere melange deux espaces sans que rien ne le
        montre, et regenerate_embeddings ne sait pas quoi refaire.
        """
        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        pl._generate_article_embeddings(law_row.id)

        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()
        assert {a.embedding_model for a in rows if a.embed} == {EMPREINTE_DOUBLURE}
        assert {a.embedding_model for a in rows if not a.embed} <= {None}

    @pytest.mark.asyncio
    async def test_overlong_article_does_not_zero_the_whole_law(
        self, db_session, law_row, _stub_service
    ):
        """
        Un seul article trop long faisait lever generate_batch_embeddings pour
        la loi ENTIERE ; l'exception etait avalee et la loi publiee sans un
        seul vecteur.
        """
        # Relatif a la limite : un texte fixe passerait sous une limite relevee
        # (78 000 caracteres pour Voyage), sans plus exercer la troncature.
        long_text = SAMPLE_TEXT + "\n\nArticle 4. Long\n" + (
            "mot " * (EmbeddingService.MAX_TEXT_LENGTH // 4 + 500)
        )
        pl._split_and_save_articles(law_row.id, long_text)

        written = pl._generate_article_embeddings(law_row.id)

        assert written > 0
        assert all(
            len(t) <= EmbeddingService.MAX_TEXT_LENGTH for t in _stub_service["texts"]
        )


class TestFtsUpdate:

    @pytest.mark.asyncio
    async def test_updates_search_vectors(self, db_session, law_row):
        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        await pl._update_fts_vectors_async(db_session, law_row.id)
        await db_session.commit()

        from sqlalchemy import text as sql_text

        result = await db_session.execute(
            sql_text(
                "SELECT count(*) FROM articles "
                "WHERE law_id = :id AND search_vector IS NOT NULL"
            ),
            {"id": law_row.id},
        )
        assert result.scalar() > 0


class TestCategoryPersistence:
    """
    Le classifieur tournait sur chaque document et son resultat etait jete.
    _classify_category ne renvoyait qu'un NOM de categorie, et
    _update_law_metadata n'avait donc rien a ecrire dans laws.category_id :
    toutes les lois restaient sans categorie et les pages de categorie du front
    etaient vides, alors que la classification avait bien eu lieu.
    """

    def test_classifier_returns_the_category_id(self, monkeypatch, classeur):
        doublure = classeur(AFFAIRES, secondaires=[FINANCES])
        rows = {name.lower(): 1000 + offset for offset, name in enumerate(CANONICAL_DOMAINS)}
        monkeypatch.setattr(pl, "SyncSessionLocal", _FakeSessionFactory(rows), raising=True)

        result = pl._classify_category(
            "Décret portant approbation des statuts de la société SOCADEL",
            "La présente loi fixe le régime des sociétés commerciales et du "
            "registre du commerce au Cameroun.",
        )

        assert doublure.titres == ["Décret portant approbation des statuts de la société SOCADEL"]
        assert result["category"] == AFFAIRES
        assert result["category_id"] == rows[AFFAIRES.lower()]
        # Les suggestions sont des IDENTIFIANTS, domaine retenu en tete.
        assert result["suggested"] == [rows[AFFAIRES.lower()], rows[FINANCES.lower()]]
        assert 0.0 <= result["confidence"] <= 1.0
        assert result["error"] is None

    def test_missing_domain_leaves_the_id_null_and_reports_it(self, monkeypatch, classeur):
        """
        Sans ligne correspondante en base, mieux vaut aucune categorie qu'une
        fausse : c'est l'invention d'un identifiant qui a produit le bug.
        """
        classeur(FINANCES)
        monkeypatch.setattr(
            pl, "SyncSessionLocal", _FakeSessionFactory({}), raising=True
        )

        result = pl._classify_category(
            "Loi portant Code Général des Impôts", "texte fiscal sur l'impôt"
        )

        assert result["category"] == FINANCES
        assert result["category_id"] is None
        assert result["suggested"] == []
        assert "absent de la table categories" in result["error"]

    def test_resolves_the_id_by_name_never_by_position(self, monkeypatch, classeur):
        """
        La table est semee dans l'ordre INVERSE : tout code qui resoudrait un
        domaine par sa position designerait le mauvais identifiant.
        """
        classeur(FONCTION_PUBLIQUE)
        rows = {
            name.lower(): 1000 + offset
            for offset, name in enumerate(reversed(CANONICAL_DOMAINS))
        }
        monkeypatch.setattr(
            pl, "SyncSessionLocal", _FakeSessionFactory(rows), raising=True
        )

        result = pl._classify_category(
            "Décret portant nomination d'un Inspecteur Général", ""
        )

        assert result["category"] == FONCTION_PUBLIQUE
        assert result["category_id"] == rows[FONCTION_PUBLIQUE.lower()]

    def test_classement_indisponible_laisse_la_categorie_nulle(self, monkeypatch):
        """
        L'ancien classifieur rendait « Droit Administratif » pendant un 429 :
        la loi semblait classee, et personne ne la reprenait.
        """
        from app.services import legal_domain_classifier
        from app.services.legal_domain_classifier import ClassementIndisponible

        class _EnPanne:
            def classify(self, *args, **kwargs):
                raise ClassementIndisponible("quota Groq", retry_after=600, quota=True)

        monkeypatch.setattr(
            legal_domain_classifier, "get_legal_domain_classifier", lambda: _EnPanne()
        )

        result = pl._classify_category("Loi de finances 2024", "texte budgetaire")

        assert result["category"] is None
        assert result["category_id"] is None
        assert result["confidence"] is None
        assert result["suggested"] == []
        assert "quota Groq" in result["error"]
        assert "reclassify_domains.py" in result["error"]

    @pytest.mark.asyncio
    async def test_metadata_update_writes_the_category(self, db_session, law_row):
        from sqlalchemy import select

        from app.models.law import Category

        # L'identifiant est LU en base par son nom. L'ecrire en dur (2) etait
        # precisement la forme du bug d'origine, et cassait des que la fixture
        # cessait d'attribuer les identifiants par position.
        target = (await db_session.execute(
            select(Category.id).where(Category.name == AFFAIRES)
        )).scalar_one()

        pl._update_law_metadata(
            law_row.id,
            language="fr",
            language_confidence=0.9,
            category=AFFAIRES,
            category_confidence=0.8,
            category_id=target,
        )

        refreshed = (await db_session.execute(
            select(Law).where(Law.id == law_row.id)
        )).scalar_one()
        await db_session.refresh(refreshed)

        assert refreshed.category_id == target
        assert refreshed.category_confidence == pytest.approx(0.8)
        assert refreshed.status == "published"


class TestPipelineSansClassement:
    """
    Le pipeline publie une loi meme sans categorie, et le dit dans
    processing_error. Les etapes lourdes sont doublees : ce qui est verifie,
    c'est ce qui part vers _update_law_metadata.
    """

    @pytest.fixture
    def etapes(self, monkeypatch):
        ecrit = {}
        monkeypatch.setattr(pl, "_detect_language", lambda text: {"language": "fr", "confidence": 0.9})
        monkeypatch.setattr(pl, "_split_and_save_articles", lambda *a, **k: 4)
        monkeypatch.setattr(pl, "_generate_article_embeddings", lambda law_id: 0)
        monkeypatch.setattr(pl, "_update_law_metadata", lambda law_id, **k: ecrit.update(k))
        return ecrit

    def test_les_deux_anomalies_sont_gardees(self, monkeypatch, etapes):
        """L'erreur des embeddings ecrasait celle du classement."""
        monkeypatch.setattr(pl, "_classify_category", lambda *a: {
            **pl._SANS_CLASSEMENT, "error": "Classement indisponible (quota Groq)",
        })

        pl._run_analysis_pipeline(1, Law(title="Loi", type="loi"), "x" * 60, file_errors=["page 3 illisible"])

        erreurs = etapes["processing_error"].split(" | ")
        assert erreurs[0] == "page 3 illisible"
        assert erreurs[1] == "Classement indisponible (quota Groq)"
        assert erreurs[2].startswith("Aucun embedding genere pour 4 articles")
        assert etapes["category_id"] is None
        assert etapes["category_confidence"] is None

    def test_classer_false_n_appelle_pas_le_classement(self, monkeypatch, etapes):
        def _interdit(*args):
            raise AssertionError("le classement devait etre differe")

        monkeypatch.setattr(pl, "_classify_category", _interdit)
        monkeypatch.setattr(pl, "_generate_article_embeddings", lambda law_id: 4)

        resultat = pl._run_analysis_pipeline(
            1, Law(title="Loi", type="loi"), "x" * 60, classer=False
        )

        assert resultat["category"] is None
        assert etapes["category"] is None
        # Differer le classement n'est pas une anomalie.
        assert etapes["processing_error"] is None


class TestChunkRefinement:
    """
    chunk_refiner est enfin cable au pipeline.
    Il existait, testé, appelé par personne. Il classe chaque chunk, decide ce
    qui merite un vecteur et prepare le texte reellement envoye au modele. Rien
    n'est supprime : un visa ou une formule d'execution reste consultable et
    cherchable en plein texte, il ne consomme simplement plus d'appel
    d'embedding et ne pollue plus les resultats semantiques.
    """

    @pytest.mark.asyncio
    async def test_classifies_every_chunk(self, db_session, law_row):
        from sqlalchemy import select

        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()

        assert rows
        assert all(a.kind for a in rows), "chaque chunk doit porter une nature"
        assert {a.kind for a in rows} & {"article", "legal_basis", "fragment"}

    @pytest.mark.asyncio
    async def test_visas_are_kept_but_not_embedded(self, db_session, law_row):
        """Le texte des visas reste consultable ; il ne part pas au vectoriel."""
        from sqlalchemy import select

        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()

        non_articles = [a for a in rows if a.kind in ("legal_basis", "preamble", "fragment")]
        assert non_articles, "le preambule et les visas doivent etre conserves"
        assert all(a.content for a in non_articles)
        assert all(a.embed is False for a in non_articles)

    @pytest.mark.asyncio
    async def test_embed_text_carries_the_document_header(self, db_session, law_row):
        """
        Sans en-tete, "Article 3.- La depense sera imputee sur le budget de
        l'Etat" est indistinguable des milliers d'articles identiques du corpus.
        """
        from sqlalchemy import select

        pl._split_and_save_articles(law_row.id, SAMPLE_TEXT)

        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()

        embeddable = [a for a in rows if a.embed]
        assert embeddable
        for article in embeddable:
            assert article.embed_text, "un chunk vectorisable doit avoir un embed_text"
            assert law_row.reference in article.embed_text
            # content reste INTACT : ce qui est affiche et cite ne change pas.
            assert article.embed_text != article.content
            assert article.content in article.embed_text

    @pytest.mark.asyncio
    async def test_markdown_emphasis_no_longer_hides_articles(self, db_session, law_row):
        """
        LlamaParse rend du markdown : "**ARTICLE 1ER**:" n'etait pas reconnu par
        les motifs d'article, qui attendent "Article" en debut de ligne. Des
        documents entiers ressortaient sans un seul article.
        """
        from sqlalchemy import select

        markdown = """
# LOI N° 2024-600 DU 2 JUIN 2024
**ARTICLE 1ER**: La présente loi fixe le régime applicable aux établissements
publics administratifs et aux entreprises du secteur public et parapublic.
**ARTICLE 2**: Les dispositions de la présente loi s'appliquent sans prejudice
des textes particuliers regissant certains etablissements.
"""
        count = pl._split_and_save_articles(law_row.id, markdown)

        rows = (await db_session.execute(
            select(Article).where(Article.law_id == law_row.id)
        )).scalars().all()

        assert count > 0
        numbers = {a.number for a in rows}
        assert "1" in numbers or "1ER" in {n.upper() for n in numbers}