veye-lalwa/tests/test_update.py

"""Run de veille : détection des écarts, classification, export.

Aucun test ne va sur le réseau : les collectes sont fabriquées à la main, ce
qui permet de vérifier exactement quels écarts le pipeline doit voir — et
surtout lesquels il ne doit pas inventer.
"""

from __future__ import annotations

import json
import sqlite3
from datetime import date

import pytest

from pipeline import db
from pipeline.collecteurs.base import ResultatCollecte, TexteCollecte
from pipeline.modeles import (
    Confiance,
    DecisionCC,
    Evenement,
    Pertinence,
    ResultatCC,
    Source,
    Statut,
    Texte,
    TypeEtape,
    TypeTexte,
)
from pipeline.update import appliquer, classer_nouveau, comparer, exporter_json


@pytest.fixture
def base(tmp_path) -> sqlite3.Connection:
    """Base minimale : une loi promulguée et une loi en attente de décision."""
    cx = db.initialiser(tmp_path / "veille.db")
    db.enregistrer_textes(
        cx,
        [
            Texte(
                id="loi-2026-491",
                numero_officiel="2026-491",
                type=TypeTexte.LOI,
                titre_court="Chlordécone : responsabilité de l'État",
                statut=Statut.PROMULGUEE,
                date_promulgation=date(2026, 6, 12),
                guadeloupe_pertinence=Pertinence.FORTE,
                sources=[
                    Source(
                        url="https://www.legifrance.gouv.fr/jorf/id/JORFTEXT000054245243",
                        editeur="Légifrance",
                    )
                ],
                evenements=[
                    Evenement(
                        date_evenement=date(2026, 6, 12),
                        type_etape=TypeEtape.PROMULGATION,
                        description="Promulgation",
                    )
                ],
            ),
            Texte(
                id="loi-riposte",
                type=TypeTexte.LOI,
                titre_court="Loi « Riposte » (ordre public, sécurité, tranquillité)",
                statut=Statut.ADOPTEE_NON_PROMULGUEE,
                date_adoption=date(2026, 7, 21),
                sources=[Source(url="https://www.assemblee-nationale.fr/dyn/17/textes/l17t0340")],
                decisions_cc=[
                    DecisionCC(
                        numero_affaire="2026-915 DC",
                        date_saisine=date(2026, 7, 24),
                        resultat=ResultatCC.EN_INSTANCE,
                    )
                ],
            ),
        ],
    )
    return cx


def _collecte(**kwargs) -> ResultatCollecte:
    return ResultatCollecte(collecteur="essai", **kwargs)


# ─────────────────────────────────────────────────────────────────────────────
# Détection des écarts
# ─────────────────────────────────────────────────────────────────────────────


def test_aucun_ecart_quand_la_source_confirme_la_base(base) -> None:
    """Le cas normal : la veille ne doit rien signaler quand rien ne bouge."""
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="LOI n° 2026-491 du 12 juin 2026 chlordécone",
                source_url="https://www.legifrance.gouv.fr/jorf/id/JORFTEXT000054245243",
                collecteur="essai",
                numero_officiel="2026-491",
                statut=Statut.PROMULGUEE,
                date_promulgation=date(2026, 6, 12),
            )
        ]
    )
    rapport = comparer(base, [collecte])
    assert rapport.ecarts == []


def test_un_texte_inconnu_est_un_ajout(base) -> None:
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="LOI n° 2026-700 du 1er septembre 2026 relative aux transports scolaires",
                source_url="https://www.legifrance.gouv.fr/jorf/id/X",
                collecteur="essai",
                numero_officiel="2026-700",
                statut=Statut.PROMULGUEE,
                date_promulgation=date(2026, 9, 1),
            )
        ]
    )
    rapport = comparer(base, [collecte])
    assert rapport.ajouts == 1
    assert rapport.ecarts[0].nature == "ajout"


def test_un_changement_de_statut_est_detecte(base) -> None:
    """La promulgation de la loi « Riposte » : l'événement que la veille attend."""
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="Loi « Riposte » (ordre public, sécurité, tranquillité)",
                source_url="https://www.legifrance.gouv.fr/jorf/id/Y",
                collecteur="essai",
                statut=Statut.PROMULGUEE,
                date_promulgation=date(2026, 8, 26),
            )
        ]
    )
    rapport = comparer(base, [collecte])
    statuts = [e for e in rapport.ecarts if e.nature == "statut"]
    assert len(statuts) == 1
    assert statuts[0].texte_id == "loi-riposte"
    assert statuts[0].ancienne_valeur == "adoptee_non_promulguee"
    assert statuts[0].nouvelle_valeur == "promulguee"


def test_une_decision_rendue_est_detectee(base) -> None:
    collecte = _collecte(
        decisions_cc=[
            DecisionCC(
                numero_affaire="2026-915 DC",
                date_decision=date(2026, 8, 24),
                resultat=ResultatCC.NON_CONFORMITE_PARTIELLE,
            )
        ]
    )
    rapport = comparer(base, [collecte])
    natures = {e.champ for e in rapport.ecarts if e.nature == "decision_cc"}
    assert any("date_decision" in (c or "") for c in natures)
    assert any("resultat" in (c or "") for c in natures)


def test_une_affaire_inconnue_est_signalee(base) -> None:
    """Cas réel : le corpus ignorait le numéro 2026-907 DC de la LPM."""
    collecte = _collecte(
        decisions_cc=[
            DecisionCC(
                numero_affaire="2026-907 DC",
                date_saisine=date(2026, 7, 6),
                resume="Loi actualisant la programmation militaire",
            )
        ]
    )
    rapport = comparer(base, [collecte])
    assert any("affaire inconnue" in e.description for e in rapport.ecarts)


def test_une_affaire_en_instance_deja_connue_ne_produit_rien(base) -> None:
    collecte = _collecte(
        decisions_cc=[
            DecisionCC(
                numero_affaire="2026-915 DC",
                date_saisine=date(2026, 7, 24),
                resultat=ResultatCC.EN_INSTANCE,
            )
        ]
    )
    assert comparer(base, [collecte]).ecarts == []


# ─────────────────────────────────────────────────────────────────────────────
# Écriture
# ─────────────────────────────────────────────────────────────────────────────


def test_le_dry_run_n_ecrit_rien(base) -> None:
    """`comparer` ne doit jamais toucher à la base."""
    avant = db.statistiques(base)
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="Loi « Riposte » (ordre public, sécurité, tranquillité)",
                source_url="https://x.fr/y",
                collecteur="essai",
                statut=Statut.PROMULGUEE,
            )
        ]
    )
    comparer(base, [collecte])
    assert db.statistiques(base) == avant


def test_appliquer_ecrit_le_changement_et_le_journalise(base) -> None:
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="Loi « Riposte » (ordre public, sécurité, tranquillité)",
                source_url="https://x.fr/y",
                collecteur="essai",
                statut=Statut.PROMULGUEE,
            )
        ]
    )
    rapport = comparer(base, [collecte])
    run_id = db.ouvrir_run(base, "update")
    appliquer(base, rapport, run_id)

    statut = base.execute("SELECT statut FROM textes WHERE id = 'loi-riposte'").fetchone()[0]
    assert statut == "promulguee"

    journal = base.execute(
        "SELECT nature, champ, nouvelle_valeur FROM veille_changements WHERE run_id = ?",
        (run_id,),
    ).fetchall()
    assert any(
        ligne["champ"] == "statut" and ligne["nouvelle_valeur"] == "promulguee"
        for ligne in journal
    )


# ─────────────────────────────────────────────────────────────────────────────
# Classification d'un texte nouveau
# ─────────────────────────────────────────────────────────────────────────────


def test_un_texte_collecte_est_marque_comme_automatique() -> None:
    """§5.3 : la machine propose avec une confiance basse, l'humain valide."""
    texte = classer_nouveau(
        TexteCollecte(
            titre="LOI n° 2026-700 du 1er septembre 2026 relative à l'eau en Guadeloupe",
            source_url="https://www.legifrance.gouv.fr/jorf/id/Z",
            collecteur="legifrance",
            numero_officiel="2026-700",
            statut=Statut.PROMULGUEE,
            date_promulgation=date(2026, 9, 1),
        )
    )
    assert texte.id == "loi-2026-700"
    assert texte.confiance is Confiance.LOW
    assert texte.a_verifier is True
    assert "déduits" in (texte.motif_verification or "")
    assert texte.guadeloupe_pertinence is Pertinence.FORTE  # « Guadeloupe » dans l'intitulé


def test_un_texte_sans_numero_recoit_un_identifiant_lisible() -> None:
    texte = classer_nouveau(
        TexteCollecte(
            titre="Proposition de loi relative aux sargasses en Martinique",
            source_url="https://www.senat.fr/x",
            collecteur="senat",
            statut=Statut.NAVETTE,
        )
    )
    assert texte.id and " " not in texte.id
    assert "sargasses" in texte.id


# ─────────────────────────────────────────────────────────────────────────────
# Export statique
# ─────────────────────────────────────────────────────────────────────────────


def test_export_json(base, tmp_path, monkeypatch) -> None:
    from pipeline import chemins

    destination = tmp_path / "textes.json"
    monkeypatch.setattr(chemins, "DUMP_JSON", destination)

    total = exporter_json(base)
    assert total == 2

    charge = json.loads(destination.read_text(encoding="utf-8"))
    assert charge["date_arrete_corpus"] == "2026-07-25"
    identifiants = {t["id"] for t in charge["textes"]}
    assert identifiants == {"loi-2026-491", "loi-riposte"}

    chlordecone = next(t for t in charge["textes"] if t["id"] == "loi-2026-491")
    assert chlordecone["sources"][0]["url"].startswith("https://www.legifrance.gouv.fr")
    assert chlordecone["evenements"][0]["type_etape"] == "promulgation"

    riposte = next(t for t in charge["textes"] if t["id"] == "loi-riposte")
    assert riposte["decisions_cc"][0]["numero_affaire"] == "2026-915 DC"


def test_rapport_lisible(base) -> None:
    rapport = comparer(base, [_collecte()])
    rapport.mode = "dry-run"
    texte = rapport.en_texte()
    assert "Rapport de veille" in texte
    assert "conforme aux sources officielles" in texte


# ─────────────────────────────────────────────────────────────────────────────
# Cotation Guadeloupe : les négations
# ─────────────────────────────────────────────────────────────────────────────


def test_une_mention_niee_ne_vaut_pas_pertinence() -> None:
    """« Sans portée pour la Guadeloupe » nomme le territoire pour l'écarter.

    Sans lecture de la négation, la PPL montagne recevait la cotation la plus
    forte alors que le rapport dit l'inverse — et remontait dans la facette
    « pertinence forte pour la Guadeloupe ».
    """
    from pipeline.guadeloupe import coter

    cotation = coter(
        "PPL « pour une montagne vivante et souveraine ». Continuité de "
        "l'urbanisation ; sans portée pour la Guadeloupe."
    )
    assert cotation.pertinence is Pertinence.FAIBLE
    assert cotation.score == 0
    assert cotation.exclusions
    assert "écarte explicitement" in (cotation.note or "")


def test_pas_de_volet_outre_mer_est_une_negation() -> None:
    from pipeline.guadeloupe import coter

    cotation = coter("PJL logement. Véto des maires ; pas de volet outre-mer à ce stade.")
    assert cotation.pertinence is Pertinence.FAIBLE
    assert cotation.score == 0


def test_une_mention_affirmee_compte_toujours() -> None:
    from pipeline.guadeloupe import coter

    cotation = coter(
        "Chlordécone : responsabilité de l'État. Guadeloupe et Martinique au cœur du texte."
    )
    assert cotation.pertinence is Pertinence.FORTE
    assert "guadeloupe" in cotation.expressions
    assert not cotation.exclusions


def test_la_negation_ne_porte_que_sur_son_voisinage() -> None:
    """Une exclusion lointaine ne doit pas annuler une mention affirmée."""
    from pipeline.guadeloupe import coter

    cotation = coter(
        "Fouilles jusqu'à 40 km du littoral, soit la quasi-totalité de la Guadeloupe. "
        + "Texte de portée générale. " * 12
        + "Le dispositif est sans objet pour les collectivités du Pacifique."
    )
    assert cotation.pertinence is Pertinence.FORTE


# ─────────────────────────────────────────────────────────────────────────────
# Application effective : créer, mettre à jour, converger
# ─────────────────────────────────────────────────────────────────────────────


def test_un_ajout_cree_reellement_le_texte(base) -> None:
    """Un écart d'ajout doit écrire le texte, pas seulement le journaliser."""
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="LOI n° 2026-103 du 19 février 2026 de finances pour 2026",
                source_url="https://www.legifrance.gouv.fr/jorf/id/JORFTEXT000999",
                collecteur="senat",
                numero_officiel="2026-103",
                statut=Statut.PROMULGUEE,
                date_promulgation=date(2026, 2, 19),
                sources=[
                    Source(
                        url="https://www.legifrance.gouv.fr/jorf/id/JORFTEXT000999",
                        editeur="Légifrance",
                    )
                ],
            )
        ]
    )
    rapport = comparer(base, [collecte])
    crees = appliquer(base, rapport, db.ouvrir_run(base, "update"))

    assert crees == 1
    ligne = base.execute("SELECT * FROM textes WHERE id = 'loi-2026-103'").fetchone()
    assert ligne is not None
    assert ligne["confiance"] == "low"
    assert ligne["a_verifier"] == 1
    assert ligne["source_seed"] == "collecteur:senat"
    # Un texte sans source serait un texte invérifiable : la collecte doit
    # transmettre la sienne.
    assert base.execute(
        "SELECT COUNT(*) FROM sources WHERE texte_id = 'loi-2026-103'"
    ).fetchone()[0] >= 1


def test_une_decision_collectee_est_ecrite_pas_seulement_journalisee(base) -> None:
    """Sans écriture, le même écart serait redétecté à chaque passage."""
    collecte = _collecte(
        decisions_cc=[
            DecisionCC(
                numero_affaire="2026-915 DC",
                date_decision=date(2026, 8, 24),
                resultat=ResultatCC.NON_CONFORMITE_PARTIELLE,
            )
        ]
    )
    rapport = comparer(base, [collecte])
    appliquer(base, rapport, db.ouvrir_run(base, "update"))

    ligne = base.execute(
        "SELECT * FROM decisions_cc WHERE numero_affaire = '2026-915 DC'"
    ).fetchone()
    assert ligne["date_decision"] == "2026-08-24"
    assert ligne["resultat"] == "non_conformite_partielle"


def test_le_pipeline_converge(base) -> None:
    """Rejouer la même collecte après application ne doit plus rien signaler.

    C'est la propriété qui rend une veille utilisable : si le rapport répétait
    les mêmes écarts à chaque passage, il deviendrait illisible et on cesserait
    de le lire.
    """
    collecte = _collecte(
        textes=[
            TexteCollecte(
                titre="LOI n° 2026-103 du 19 février 2026 de finances pour 2026",
                source_url="https://www.legifrance.gouv.fr/jorf/id/JORFTEXT000999",
                collecteur="senat",
                numero_officiel="2026-103",
                statut=Statut.PROMULGUEE,
                date_promulgation=date(2026, 2, 19),
            )
        ],
        decisions_cc=[
            DecisionCC(
                numero_affaire="2026-915 DC",
                date_decision=date(2026, 8, 24),
                resultat=ResultatCC.NON_CONFORMITE_PARTIELLE,
            )
        ],
    )

    premier = comparer(base, [collecte])
    assert premier.ecarts, "le premier passage doit voir des écarts"
    appliquer(base, premier, db.ouvrir_run(base, "update"))

    second = comparer(base, [collecte])
    assert second.ecarts == [], "le second passage ne doit plus rien signaler"


def test_une_affaire_orpheline_est_rattachee_par_son_intitule(base) -> None:
    """Cas réel : le Conseil donne l'intitulé complet, la base le nom d'usage."""
    collecte = _collecte(
        decisions_cc=[
            DecisionCC(
                numero_affaire="2026-916 DC",
                date_saisine=date(2026, 8, 1),
                resume="Loi « Riposte » ordre public sécurité tranquillité",
            )
        ]
    )
    rapport = comparer(base, [collecte])
    (ecart,) = [e for e in rapport.ecarts if e.nature == "decision_cc"]
    assert ecart.texte_id == "loi-riposte"
    assert "rattachée à loi-riposte" in ecart.description