Bertrand Benjamin
a1da4fc97f
Refuser une modification, c'est reenvoyer la valeur actuelle du document. Mais
l'enregistrement ramene les donnees a une forme canonique (aujourd'hui les
numeros de lot) : sur ces champs, le refus n'est pas conserve. L'interface le
laissait croire.
Plutot que de traiter le cas des lots en dur, le serveur devient seul juge de ce
qu'il stocke et l'annonce : `canonicalize_extraction` regroupe les reecritures
faites a la persistance, `save_document` et l'apercu passent par elle, et
/api/documents/{id}/re-extract renvoie `previous_canonical` a cote de
`previous_data`. Le front compare ces deux jeux et signale, sous chaque case
decochee, la valeur que l'enregistrement reecrira — sans connaitre aucune regle.
Une transformation ajoutee plus tard remontera donc toute seule dans l'interface.
Toutes les differences restent visibles et toutes les cases actives : voir ce
qui change et pouvoir se prononcer prime, quitte a etre prevenu que ce champ-la
reviendra reecrit.
Supprime au passage frontend/src/utils/lots.js, qui dupliquait la regle de
normalisation cote client.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
193 lines
7.1 KiB
Python
193 lines
7.1 KiB
Python
"""Tests du remplacement d'un document par son ID (re-extraction validee)."""
|
|
|
|
import copy
|
|
import json
|
|
from datetime import date
|
|
|
|
import pytest
|
|
|
|
from plesna_gerance.database import storage
|
|
from plesna_gerance.database.models import Depense, Document, Revenu, Tag
|
|
from plesna_gerance.database.service import DatabaseService, DuplicateDocumentError
|
|
|
|
|
|
@pytest.fixture
|
|
def pdf_bytes():
|
|
"""Contenu binaire arbitraire : le stockage ne relit jamais le PDF ici."""
|
|
return b"%PDF-1.4 fake"
|
|
|
|
|
|
def test_replace_document_met_a_jour_meme_si_la_reference_change(
|
|
db_session, sample_data
|
|
):
|
|
"""Une re-extraction qui corrige la reference met a jour le meme document.
|
|
|
|
C'est tout l'interet du remplacement par ID : avec un enregistrement par
|
|
(reference, date), ce cas creerait un second document.
|
|
"""
|
|
service = DatabaseService(db_session)
|
|
original = service.save_document(data=sample_data)
|
|
original_id = original.id
|
|
|
|
new_data = copy.deepcopy(sample_data)
|
|
new_data["metadata"]["document"]["reference"] = "REF001-CORRIGEE"
|
|
|
|
updated = service.save_document(data=new_data, replace_document_id=original_id)
|
|
|
|
assert db_session.query(Document).count() == 1
|
|
assert updated.reference == "REF001-CORRIGEE"
|
|
assert updated.date == date(2024, 1, 15)
|
|
# L'ID survit au remplacement : les liens vers le document restent valides.
|
|
assert updated.id == original_id
|
|
|
|
|
|
def test_replace_document_regenere_les_donnees_derivees(db_session, sample_data):
|
|
"""Revenus et depenses sont remplaces, pas cumules."""
|
|
service = DatabaseService(db_session)
|
|
original = service.save_document(data=sample_data)
|
|
|
|
new_data = copy.deepcopy(sample_data)
|
|
new_data["recapitulatif_operations"][0]["montants"]["debit"] = 90.0
|
|
new_data["situation_locataires"][0]["lignes"][0]["loyers"] = 600.0
|
|
service.save_document(data=new_data, replace_document_id=original.id)
|
|
|
|
depenses = db_session.query(Depense).filter_by(document_id=original.id).all()
|
|
revenus = db_session.query(Revenu).filter_by(document_id=original.id).all()
|
|
assert len(depenses) == 1 and depenses[0].debit == 90.0
|
|
assert len(revenus) == 1 and revenus[0].loyers == 600.0
|
|
|
|
|
|
def test_replace_document_conserve_le_pdf_et_reecrit_le_json(
|
|
db_session, sample_data, pdf_bytes
|
|
):
|
|
service = DatabaseService(db_session)
|
|
original = service.save_document(
|
|
data=sample_data, source_file="cr.pdf", pdf_content=pdf_bytes
|
|
)
|
|
pdf_path, json_path = original.pdf_path, original.json_path
|
|
assert pdf_path and json_path
|
|
|
|
new_data = copy.deepcopy(sample_data)
|
|
new_data["situation_locataires"][0]["locataire"]["nom"] = "MARTIN"
|
|
|
|
# Re-enregistrement sans PDF : celui deja stocke doit etre conserve.
|
|
updated = service.save_document(
|
|
data=new_data, source_file="cr.pdf", replace_document_id=original.id
|
|
)
|
|
|
|
assert updated.pdf_path == pdf_path
|
|
assert updated.json_path == json_path
|
|
assert storage.read_pdf(pdf_path) == pdf_bytes
|
|
stored = storage.read_json(json_path)
|
|
assert stored["situation_locataires"][0]["locataire"]["nom"] == "MARTIN"
|
|
|
|
|
|
def test_replace_document_refuse_une_collision_avec_un_autre_document(
|
|
db_session, sample_data
|
|
):
|
|
"""La nouvelle extraction ne doit pas ecraser un document voisin."""
|
|
service = DatabaseService(db_session)
|
|
premier = service.save_document(data=sample_data)
|
|
|
|
autre_data = copy.deepcopy(sample_data)
|
|
autre_data["metadata"]["document"]["date"] = "2024-02-15"
|
|
service.save_document(data=autre_data)
|
|
|
|
# Le premier document re-extrait porterait la (reference, date) du second.
|
|
collision_data = copy.deepcopy(autre_data)
|
|
with pytest.raises(DuplicateDocumentError):
|
|
service.save_document(data=collision_data, replace_document_id=premier.id)
|
|
|
|
assert db_session.query(Document).count() == 2
|
|
|
|
|
|
def test_replace_document_inconnu_leve_value_error(db_session, sample_data):
|
|
service = DatabaseService(db_session)
|
|
with pytest.raises(ValueError):
|
|
service.save_document(data=sample_data, replace_document_id=4242)
|
|
|
|
|
|
def test_un_numero_de_lot_refuse_est_quand_meme_reecrit(db_session, sample_data):
|
|
"""L'enregistrement ramene les donnees a leur forme canonique.
|
|
|
|
Refuser une reecriture de numero de lot ("0001" -> "01") ne la conserve donc
|
|
pas : c'est exactement ce que `previous_canonical` sert a annoncer avant
|
|
l'enregistrement.
|
|
"""
|
|
service = DatabaseService(db_session)
|
|
original = service.save_document(data=copy.deepcopy(sample_data))
|
|
|
|
refus = copy.deepcopy(sample_data)
|
|
refus["situation_locataires"][0]["lot"]["numero"] = "0001"
|
|
updated = service.save_document(data=refus, replace_document_id=original.id)
|
|
|
|
stored = json.loads(updated.json_data)
|
|
assert stored["situation_locataires"][0]["lot"]["numero"] == "01"
|
|
|
|
|
|
def test_re_extract_annonce_les_reecritures_de_l_enregistrement(
|
|
db_session, sample_data, pdf_bytes, monkeypatch
|
|
):
|
|
"""`previous_canonical` montre ce que l'enregistrement imposerait.
|
|
|
|
L'interface s'en sert pour prevenir, avant que l'utilisateur ne se prononce,
|
|
qu'un refus sur ce champ ne serait pas conserve — sans connaitre la regle.
|
|
"""
|
|
from plesna_gerance.api.routes import documents as documents_routes
|
|
|
|
service = DatabaseService(db_session)
|
|
document = service.save_document(
|
|
data=copy.deepcopy(sample_data), source_file="cr.pdf", pdf_content=pdf_bytes
|
|
)
|
|
|
|
# JSON herite, anterieur a la normalisation des lots.
|
|
ancien = copy.deepcopy(sample_data)
|
|
ancien["situation_locataires"][0]["lot"]["numero"] = "0001"
|
|
document.json_data = json.dumps(ancien, ensure_ascii=False)
|
|
db_session.commit()
|
|
|
|
monkeypatch.setattr(
|
|
documents_routes,
|
|
"extract_compte_rendu",
|
|
lambda path: copy.deepcopy(sample_data),
|
|
)
|
|
result = documents_routes.re_extract_document(document.id, db_session)
|
|
|
|
# Les donnees actuelles sont renvoyees telles quelles...
|
|
assert result["previous_data"]["situation_locataires"][0]["lot"]["numero"] == "0001"
|
|
# ...et leur forme canonique montre la reecriture a venir.
|
|
assert (
|
|
result["previous_canonical"]["situation_locataires"][0]["lot"]["numero"] == "01"
|
|
)
|
|
|
|
|
|
def test_get_depenses_tags_puis_report_sur_la_nouvelle_extraction(
|
|
db_session, sample_data
|
|
):
|
|
"""Les tags actuels sont indexes comme les operations, et reportables."""
|
|
service = DatabaseService(db_session)
|
|
tag = Tag(nom="ENTRETIEN")
|
|
db_session.add(tag)
|
|
db_session.flush()
|
|
|
|
original = service.save_document(
|
|
data=sample_data, depenses_tags=[{"index": 0, "tag_id": tag.id}]
|
|
)
|
|
|
|
tags = service.get_depenses_tags(original.id)
|
|
assert tags == [{"index": 0, "tag_id": tag.id, "tag_nom": "ENTRETIEN"}]
|
|
|
|
# Report sur une nouvelle extraction du meme document.
|
|
new_data = copy.deepcopy(sample_data)
|
|
new_data["recapitulatif_operations"][0]["montants"]["debit"] = 75.0
|
|
updated = service.save_document(
|
|
data=new_data,
|
|
depenses_tags=[{"index": t["index"], "tag_id": t["tag_id"]} for t in tags],
|
|
replace_document_id=original.id,
|
|
)
|
|
|
|
depenses = db_session.query(Depense).filter_by(document_id=updated.id).all()
|
|
assert len(depenses) == 1
|
|
assert depenses[0].debit == 75.0
|
|
assert depenses[0].tag_id == tag.id
|