Bertrand Benjamin
6f2530014f
Le remplacement par ID mettait a jour le document en le supprimant puis en le recreant : la ligne changeait d'ID, cassant les liens qui la referencent (URL d'edition, telechargement du PDF) juste apres un balayage de re-extraction. Le document est desormais mis a jour sur place, seules les donnees derivees (revenus, depenses) etant regenerees. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
138 lines
5.0 KiB
Python
138 lines
5.0 KiB
Python
"""Tests du remplacement d'un document par son ID (re-extraction validee)."""
|
|
|
|
import copy
|
|
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_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
|