Spaces:
Runtime error
Runtime error
| """ | |
| Le reclassement par lots : journal reprenable, quota, application relue. | |
| Le classifieur est double : ces tests portent sur ce que le script fait des | |
| verdicts — journal, reprise sans rappel, arret propre sur quota, rien | |
| d'applique sous le seuil, rien d'ecrit en simulation. | |
| Usage: | |
| pytest tests/test_scripts/test_reclassify_domains.py -v | |
| """ | |
| import csv | |
| import json | |
| import pytest | |
| from sqlalchemy import text | |
| from app.models.law import Category, Law | |
| from app.services.legal_domain_classifier import ( | |
| ADMINISTRATIF, | |
| CANONICAL_DOMAINS, | |
| FINANCES, | |
| FONCTION_PUBLIQUE, | |
| INTERNATIONAL, | |
| ClassementIndisponible, | |
| DomainResult, | |
| ResultatLot, | |
| ) | |
| from scripts.maintenance import reclassify_domains as script | |
| class _Classeur: | |
| """Rend, dans l'ordre, les verdicts programmes (un domaine, None ou une exception).""" | |
| modele = "doublure" | |
| def __init__(self, *reponses): | |
| self.reponses = list(reponses) | |
| self.lots = [] | |
| def classer_lot(self, documents, *, attente_max=None): | |
| self.lots.append(list(documents)) | |
| reponse = self.reponses.pop(0) | |
| if isinstance(reponse, Exception): | |
| raise reponse | |
| verdicts = [ | |
| None if domaine is None else DomainResult( | |
| domain=domaine, confidence=confiance, rule="groq:doublure", source="groq", | |
| runners_up=((INTERNATIONAL, 0.5),), | |
| ) | |
| for domaine, confiance in reponse | |
| ] | |
| return ResultatLot(verdicts, jetons=300, requetes=1, modele="doublure") | |
| def base(sync_db_session): | |
| """Les 14 domaines et 5 lois publiees, toutes rangees en Administratif.""" | |
| for ordre, nom in enumerate(CANONICAL_DOMAINS): | |
| sync_db_session.add(Category(name=nom, description=nom, display_order=ordre)) | |
| sync_db_session.commit() | |
| ids = dict(sync_db_session.execute(text("SELECT name, id FROM categories")).all()) | |
| for i in range(1, 6): | |
| sync_db_session.add(Law( | |
| reference=f"R-{i}", title=f"Décret n°{i}", type="decret", content="Contenu.", | |
| language="fr", status="published", category_id=ids[ADMINISTRATIF], | |
| )) | |
| sync_db_session.commit() | |
| lois = [loi.id for loi in sync_db_session.query(Law).order_by(Law.id)] | |
| return sync_db_session, ids, lois | |
| def _journal(chemin): | |
| return [json.loads(ligne) for ligne in chemin.read_text(encoding="utf-8").splitlines()] | |
| class TestClasser: | |
| def test_ecrit_un_verdict_par_loi(self, base, tmp_path): | |
| session, ids, lois = base | |
| journal = tmp_path / "verdicts.jsonl" | |
| classeur = _Classeur( | |
| [(FINANCES, 0.9), (FONCTION_PUBLIQUE, 0.8), (None, None)], | |
| [(FINANCES, 0.4), (ADMINISTRATIF, 0.7)], | |
| ) | |
| code = script.classer(session, classeur, journal=journal, taille_lot=3) | |
| assert code == 0 | |
| assert [len(lot) for lot in classeur.lots] == [3, 2] | |
| lignes = _journal(journal) | |
| assert [ligne["law_id"] for ligne in lignes] == lois | |
| assert lignes[0]["domaine"] == FINANCES | |
| assert lignes[0]["secondaires"] == [INTERNATIONAL] | |
| assert lignes[2]["statut"] == "a_revoir" | |
| assert lignes[2]["domaine"] is None | |
| assert {ligne["consignes"] for ligne in lignes} == {script.VERSION_DES_CONSIGNES} | |
| # Classer n'ecrit rien en base. | |
| assert session.query(Law).filter(Law.category_id == ids[ADMINISTRATIF]).count() == 5 | |
| def test_reprise_sans_rappel(self, base, tmp_path): | |
| session, ids, lois = base | |
| journal = tmp_path / "verdicts.jsonl" | |
| script.classer( | |
| session, _Classeur([(FINANCES, 0.9)] * 5), journal=journal, taille_lot=5 | |
| ) | |
| relance = _Classeur() | |
| code = script.classer(session, relance, journal=journal, taille_lot=5) | |
| assert code == 0 | |
| assert relance.lots == [] | |
| def test_quota_arret_propre_code_4(self, base, tmp_path): | |
| session, ids, lois = base | |
| journal = tmp_path / "verdicts.jsonl" | |
| classeur = _Classeur( | |
| [(FINANCES, 0.9), (FINANCES, 0.9)], | |
| ClassementIndisponible("TPD", retry_after=3600, quota=True), | |
| ) | |
| code = script.classer(session, classeur, journal=journal, taille_lot=2) | |
| assert code == script.SORTIE_QUOTA | |
| # Le lot deja classe est garde : la relance repart de la troisieme loi. | |
| assert [ligne["law_id"] for ligne in _journal(journal)] == lois[:2] | |
| relance = _Classeur([(FINANCES, 0.9)] * 3) | |
| script.classer(session, relance, journal=journal, taille_lot=5) | |
| assert [d.titre for d in relance.lots[0]] == ["Décret n°3", "Décret n°4", "Décret n°5"] | |
| def test_saturation_courte_attendue_sur_place(self, base, tmp_path): | |
| session, ids, lois = base | |
| sommeils = [] | |
| classeur = _Classeur( | |
| ClassementIndisponible("RPM", retry_after=20), | |
| [(FINANCES, 0.9)] * 5, | |
| ) | |
| code = script.classer( | |
| session, classeur, journal=tmp_path / "v.jsonl", taille_lot=5, dormir=sommeils.append | |
| ) | |
| assert code == 0 | |
| assert sommeils == [20] | |
| def test_panne_code_3(self, base, tmp_path): | |
| session, ids, lois = base | |
| classeur = _Classeur(ClassementIndisponible("503")) | |
| code = script.classer(session, classeur, journal=tmp_path / "v.jsonl", taille_lot=5) | |
| assert code == script.SORTIE_PANNE | |
| def test_incertains_seulement_avec_extrait(self, base, tmp_path): | |
| session, ids, lois = base | |
| journal = tmp_path / "verdicts.jsonl" | |
| session.execute( | |
| text("INSERT INTO articles (law_id, number, content, \"order\", kind) " | |
| "VALUES (:id, '1', 'Article 1er.- Est ratifié l''accord de prêt.', 1, 'article')"), | |
| {"id": lois[1]}, | |
| ) | |
| session.commit() | |
| script.classer( | |
| session, | |
| _Classeur([(FINANCES, 0.9), (INTERNATIONAL, 0.4), (None, None), (FINANCES, 0.95), (FINANCES, 0.9)]), | |
| journal=journal, taille_lot=5, | |
| ) | |
| seconde = _Classeur([(FINANCES, 0.85), (FONCTION_PUBLIQUE, 0.8)]) | |
| script.classer( | |
| session, seconde, journal=journal, taille_lot=15, | |
| longueur_extrait=500, incertains=True, | |
| ) | |
| assert [d.titre for d in seconde.lots[0]] == ["Décret n°2", "Décret n°3"] | |
| assert seconde.lots[0][0].extrait == "Article 1er.- Est ratifié l'accord de prêt." | |
| derniers = script.charger_verdicts(journal) | |
| assert derniers[lois[1]]["domaine"] == FINANCES | |
| assert derniers[lois[1]]["extrait"] == 500 | |
| # Relance apres un arret : la loi 3, encore « a revoir » mais deja | |
| # retentee avec un extrait, n'est pas repayee. | |
| troisieme = _Classeur() | |
| script.classer( | |
| session, troisieme, journal=journal, taille_lot=15, | |
| longueur_extrait=500, incertains=True, | |
| ) | |
| assert troisieme.lots == [] | |
| def test_une_ligne_coupee_n_absorbe_pas_la_suivante(self, base, tmp_path): | |
| session, ids, lois = base | |
| journal = tmp_path / "verdicts.jsonl" | |
| journal.write_text('{"law_id": 999, "titre": "Déc', encoding="utf-8") | |
| script.classer(session, _Classeur([(FINANCES, 0.9)] * 5), journal=journal, taille_lot=5) | |
| assert set(script.charger_verdicts(journal)) == set(lois) | |
| def test_le_verdict_d_une_autre_loi_ne_compte_pas(self, base, tmp_path): | |
| """Base reconstruite : l'identifiant 1 designe maintenant une autre loi.""" | |
| session, ids, lois = base | |
| journal = tmp_path / "verdicts.jsonl" | |
| journal.write_text(json.dumps({ | |
| "law_id": lois[0], "reference": "AUTRE-LOI", "statut": "ok", | |
| "domaine": FINANCES, "confiance": 0.9, "consignes": script.VERSION_DES_CONSIGNES, | |
| }) + "\n", encoding="utf-8") | |
| classeur = _Classeur([(FINANCES, 0.9)] * 5) | |
| script.classer(session, classeur, journal=journal, taille_lot=5) | |
| assert len(classeur.lots[0]) == 5 | |
| assert script.charger_verdicts(journal)[lois[0]]["reference"] == "R-1" | |
| def _verdicts(lois, *domaines): | |
| """{law_id: verdict} : (domaine, confiance) ou None pour « a revoir ».""" | |
| verdicts = {} | |
| for loi, verdict in zip(lois, domaines): | |
| domaine, confiance = verdict if verdict else (None, None) | |
| verdicts[loi] = { | |
| "law_id": loi, "statut": "ok" if verdict else "a_revoir", "domaine": domaine, | |
| "secondaires": [INTERNATIONAL] if verdict else [], "confiance": confiance, | |
| } | |
| return verdicts | |
| def _csv(chemin): | |
| with chemin.open(encoding="utf-8") as fichier: | |
| return list(csv.DictReader(fichier)) | |
| class TestAppliquer: | |
| def test_dry_run_n_ecrit_rien_en_base(self, base, tmp_path): | |
| session, ids, lois = base | |
| verdicts = _verdicts(lois, (FINANCES, 0.9), (FONCTION_PUBLIQUE, 0.8)) | |
| bilan = script.appliquer(session, verdicts, dossier=tmp_path, force=True, dry_run=True) | |
| session.expire_all() | |
| assert bilan.changees == 2 | |
| assert session.query(Law).filter(Law.category_id == ids[ADMINISTRATIF]).count() == 5 | |
| assert [ligne["apres"] for ligne in _csv(tmp_path / "changements.csv")] == [ | |
| FINANCES, FONCTION_PUBLIQUE, | |
| ] | |
| def test_la_simulation_sans_force_montre_les_categories_qu_il_remplacera(self, base, tmp_path): | |
| """ | |
| Le cas du corpus : toutes les lois ont deja une categorie. Une | |
| simulation qui n'en listait aucune laissait relire un fichier vide | |
| avant un --force qui les reecrivait toutes. | |
| """ | |
| session, ids, lois = base | |
| verdicts = _verdicts(lois, (FINANCES, 0.9)) | |
| bilan = script.appliquer(session, verdicts, dossier=tmp_path, dry_run=True) | |
| assert bilan.protegees == 1 | |
| lignes = _csv(tmp_path / "changements.csv") | |
| assert [(ligne["apres"], ligne["suite"]) for ligne in lignes] == [ | |
| (FINANCES, "gardee (--force pour appliquer)"), | |
| ] | |
| def test_l_anomalie_de_classement_est_effacee(self, base, tmp_path): | |
| session, ids, lois = base | |
| session.execute(text("UPDATE laws SET processing_error = :e WHERE id = :id"), { | |
| "id": lois[0], | |
| "e": "page 3 illisible | Classement indisponible (quota Groq) : relancer " | |
| "scripts/maintenance/reclassify_domains.py", | |
| }) | |
| session.execute(text("UPDATE laws SET processing_error = :e WHERE id = :id"), { | |
| "id": lois[1], "e": "Classement indisponible (panne) : relancer ...", | |
| }) | |
| session.commit() | |
| script.appliquer( | |
| session, _verdicts(lois, (FINANCES, 0.9), (FINANCES, 0.9)), dossier=tmp_path, force=True | |
| ) | |
| session.expire_all() | |
| assert session.get(Law, lois[0]).processing_error == "page 3 illisible" | |
| assert session.get(Law, lois[1]).processing_error is None | |
| def test_commit_toutes_les_n_lois_ecrites(self, base, tmp_path, monkeypatch): | |
| """Compter les positions sautait le commit quand la n-ieme loi etait a revoir.""" | |
| session, ids, lois = base | |
| monkeypatch.setattr(script, "LOIS_PAR_COMMIT", 2) | |
| commits = [] | |
| vrai_commit = session.commit | |
| monkeypatch.setattr(session, "commit", lambda: (commits.append(1), vrai_commit())) | |
| verdicts = _verdicts(lois, (FINANCES, 0.9), None, (FINANCES, 0.9), None, (FINANCES, 0.9)) | |
| script.appliquer(session, verdicts, dossier=tmp_path, force=True) | |
| # 3 lois ecrites : un commit a la 2e, puis le commit final. | |
| assert len(commits) == 2 | |
| def test_le_verdict_d_une_autre_loi_n_est_pas_applique(self, base, tmp_path): | |
| session, ids, lois = base | |
| verdicts = _verdicts(lois, (FINANCES, 0.9)) | |
| verdicts[lois[0]]["reference"] = "AUTRE-LOI" | |
| bilan = script.appliquer(session, verdicts, dossier=tmp_path, force=True) | |
| session.expire_all() | |
| assert bilan.a_revoir == 1 | |
| assert session.get(Law, lois[0]).category_id == ids[ADMINISTRATIF] | |
| def test_force_applique_categorie_confiance_et_suggestions(self, base, tmp_path): | |
| session, ids, lois = base | |
| verdicts = _verdicts(lois, (FINANCES, 0.9)) | |
| script.appliquer(session, verdicts, dossier=tmp_path, force=True) | |
| session.expire_all() | |
| loi = session.get(Law, lois[0]) | |
| assert loi.category_id == ids[FINANCES] | |
| assert loi.category_confidence == pytest.approx(0.9) | |
| assert loi.suggested_categories == [ids[FINANCES], ids[INTERNATIONAL]] | |
| def test_sans_force_une_categorie_posee_est_gardee(self, base, tmp_path): | |
| session, ids, lois = base | |
| session.execute(text("UPDATE laws SET category_id = NULL WHERE id = :id"), {"id": lois[1]}) | |
| session.commit() | |
| verdicts = _verdicts(lois, (FINANCES, 0.9), (FONCTION_PUBLIQUE, 0.9)) | |
| bilan = script.appliquer(session, verdicts, dossier=tmp_path) | |
| session.expire_all() | |
| assert bilan.protegees == 1 | |
| assert session.get(Law, lois[0]).category_id == ids[ADMINISTRATIF] | |
| assert session.get(Law, lois[1]).category_id == ids[FONCTION_PUBLIQUE] | |
| def test_a_revoir_et_sous_le_seuil_non_appliques(self, base, tmp_path): | |
| """Jamais d'Administratif par defaut, ni d'autre domaine invente.""" | |
| session, ids, lois = base | |
| session.execute(text("UPDATE laws SET category_id = NULL")) | |
| session.commit() | |
| verdicts = _verdicts(lois, None, (FINANCES, 0.5), (FINANCES, 0.6)) | |
| bilan = script.appliquer(session, verdicts, dossier=tmp_path, seuil=0.6, force=True) | |
| session.expire_all() | |
| assert bilan.a_revoir == 2 | |
| assert session.get(Law, lois[0]).category_id is None | |
| assert session.get(Law, lois[1]).category_id is None | |
| assert session.get(Law, lois[2]).category_id == ids[FINANCES] | |
| revoir = _csv(tmp_path / "a_revoir.csv") | |
| assert [int(ligne["law_id"]) for ligne in revoir] == lois[:2] | |
| assert revoir[0]["statut"] == "a_revoir" | |
| class TestJournal: | |
| def test_le_dernier_verdict_l_emporte_et_une_ligne_tronquee_est_ignoree(self, tmp_path): | |
| journal = tmp_path / "verdicts.jsonl" | |
| journal.write_text( | |
| '{"law_id": 1, "statut": "ok", "domaine": "A"}\n' | |
| '{"law_id": 1, "statut": "ok", "domaine": "B"}\n' | |
| '{"law_id": 2, "sta\n', | |
| encoding="utf-8", | |
| ) | |
| assert script.charger_verdicts(journal) == { | |
| 1: {"law_id": 1, "statut": "ok", "domaine": "B"} | |
| } | |
| def test_journal_absent(self, tmp_path): | |
| assert script.charger_verdicts(tmp_path / "absent.jsonl") == {} | |