feat: durcit l'exécution SQL IA, borne les uploads et ajoute des tests

- sql_executor: remplace le filtre regex fragile par une autorisation SQLite
  (set_authorizer) en complément de mode=ro ; rejette les instructions multiples
- uploads: lecture bornée des PDF (helper read_upload_limited, limite 20 Mo,
  HTTP 413) branchée sur /extract et /save-with-pdf
- tests: suite pytest (54 tests) couvrant amounts, dates, storage, sql_executor,
  uploads et DatabaseService.save_document ; pytest ajouté en dépendance dev

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-11 10:11:07 +02:00
parent c15b65f0d9
commit 635a094591
13 changed files with 646 additions and 22 deletions

View File

@@ -0,0 +1,103 @@
"""Tests de DatabaseService.save_document (logique métier centrale)."""
from datetime import date
import pytest
from plesna_gerance.database.service import DatabaseService, DuplicateDocumentError
from plesna_gerance.database.models import (
Document,
Immeuble,
Lot,
Locataire,
Revenu,
Depense,
Tag,
)
def test_save_document_creates_full_graph(db_session, sample_data):
service = DatabaseService(db_session)
doc = service.save_document(data=sample_data, source_file="cr.pdf")
assert doc.id is not None
assert doc.reference == "REF001"
assert doc.date == date(2024, 1, 15)
assert doc.solde_montant == 100.0
# Immeuble / lot / locataire créés
assert db_session.query(Immeuble).filter_by(code="IMM1").count() == 1
assert db_session.query(Lot).filter_by(numero="001").count() == 1
assert db_session.query(Locataire).filter_by(nom="DUPONT").count() == 1
# Revenu et dépense rattachés
revenus = db_session.query(Revenu).all()
assert len(revenus) == 1 and revenus[0].loyers == 500.0
depenses = db_session.query(Depense).all()
assert len(depenses) == 1 and depenses[0].debit == 50.0
def test_save_document_requires_reference_and_date(db_session):
service = DatabaseService(db_session)
with pytest.raises(ValueError):
service.save_document(data={"metadata": {"document": {}}})
def test_save_document_duplicate_raises(db_session, sample_data):
service = DatabaseService(db_session)
service.save_document(data=sample_data)
with pytest.raises(DuplicateDocumentError):
service.save_document(data=sample_data)
def test_save_document_overwrite_replaces(db_session, sample_data):
service = DatabaseService(db_session)
first = service.save_document(data=sample_data)
first_id = first.id
# Réécriture : l'ancien document est supprimé puis recréé (cascade incluse).
# NB : SQLite peut réutiliser le même rowid, on ne compare donc pas les ids.
assert first_id is not None
service.save_document(data=sample_data, overwrite=True)
assert db_session.query(Document).count() == 1
assert db_session.query(Revenu).count() == 1
assert db_session.query(Depense).count() == 1
def test_save_document_reuses_immeuble(db_session, sample_data):
service = DatabaseService(db_session)
service.save_document(data=sample_data)
# Deuxième document, même immeuble, référence différente
data2 = {**sample_data}
data2["metadata"] = {
**sample_data["metadata"],
"document": {**sample_data["metadata"]["document"], "reference": "REF002"},
}
service.save_document(data=data2)
assert db_session.query(Immeuble).filter_by(code="IMM1").count() == 1
assert db_session.query(Document).count() == 2
def test_save_document_assigns_tags(db_session, sample_data):
service = DatabaseService(db_session)
tag = db_session.query(Tag).first()
assert tag is not None
doc = service.save_document(
data=sample_data,
depenses_tags=[{"index": 0, "tag_id": tag.id}],
)
depense = db_session.query(Depense).filter_by(document_id=doc.id).first()
assert depense.tag_id == tag.id
def test_check_duplicate(db_session, sample_data):
service = DatabaseService(db_session)
assert service.check_duplicate("REF001", date(2024, 1, 15)) is None
service.save_document(data=sample_data)
found = service.check_duplicate("REF001", date(2024, 1, 15))
assert found is not None and found.reference == "REF001"