diff --git a/src/plesna_gerance/api/routes/revenus.py b/src/plesna_gerance/api/routes/revenus.py index 1d5f9fb..2afb3d5 100644 --- a/src/plesna_gerance/api/routes/revenus.py +++ b/src/plesna_gerance/api/routes/revenus.py @@ -127,6 +127,65 @@ class RevenusSummaryResponse(BaseModel): top_impayes: list[RevenuByLocataire] +# ============================================================ +# Agregats par immeuble +# ============================================================ + + +def _revenus_par_immeuble(): + """Sous-requete : montants des revenus agreges par immeuble. + + Agreger AVANT de joindre les locataires est ce qui garantit la justesse des + totaux : un lot relouee plusieurs fois a plusieurs locataires, et une + jointure directe revenus x locataires compterait chaque revenu autant de + fois qu'il y a eu d'occupants. + """ + return ( + select( + Lot.immeuble_id.label("immeuble_id"), + func.sum(Revenu.total).label("total_revenus"), + func.sum(Revenu.regles).label("total_regles"), + func.sum(Revenu.impayes).label("total_impayes"), + ) + .join(Revenu, Revenu.lot_id == Lot.id) + .group_by(Lot.immeuble_id) + .subquery() + ) + + +def _effectifs_par_immeuble(): + """Sous-requete : nombre de lots et de locataires par immeuble.""" + return ( + select( + Lot.immeuble_id.label("immeuble_id"), + func.count(func.distinct(Lot.id)).label("nb_lots"), + func.count(func.distinct(Locataire.id)).label("nb_locataires"), + ) + .outerjoin(Locataire, Locataire.lot_id == Lot.id) + .group_by(Lot.immeuble_id) + .subquery() + ) + + +def _immeuble_response(row) -> RevenuByImmeuble: + """Construit la reponse d'un immeuble a partir d'une ligne agregee.""" + revenus = row.total_revenus or 0.0 + regles = row.total_regles or 0.0 + taux = (regles / revenus * 100) if revenus > 0 else 100.0 + return RevenuByImmeuble( + immeuble_id=row.id, + immeuble_code=row.code, + adresse=row.adresse, + ville=row.ville, + nb_lots=row.nb_lots or 0, + nb_locataires=row.nb_locataires or 0, + total_revenus=revenus, + total_regles=regles, + total_impayes=row.total_impayes or 0.0, + taux_recouvrement=round(taux, 1), + ) + + # ============================================================ # Endpoints # ============================================================ @@ -241,47 +300,30 @@ async def get_revenus_summary( ) # ========== By Immeuble ========== + # Seuls les immeubles portant des revenus apparaissent dans ce resume. + montants = _revenus_par_immeuble() + effectifs = _effectifs_par_immeuble() immeuble_stmt = ( select( Immeuble.id, Immeuble.code, Immeuble.adresse, Immeuble.ville, - func.count(func.distinct(Lot.id)).label("nb_lots"), - func.count(func.distinct(Locataire.id)).label("nb_locataires"), - func.sum(Revenu.total).label("total_revenus"), - func.sum(Revenu.regles).label("total_regles"), - func.sum(Revenu.impayes).label("total_impayes"), + effectifs.c.nb_lots, + effectifs.c.nb_locataires, + montants.c.total_revenus, + montants.c.total_regles, + montants.c.total_impayes, ) - .join(Lot, Lot.immeuble_id == Immeuble.id) - .join(Revenu, Revenu.lot_id == Lot.id) - .outerjoin(Locataire, Locataire.lot_id == Lot.id) - .group_by(Immeuble.id) - .order_by(desc("total_revenus")) + .join(montants, montants.c.immeuble_id == Immeuble.id) + .outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id) + .order_by(desc(montants.c.total_revenus)) ) if immeuble_id: immeuble_stmt = immeuble_stmt.where(Immeuble.id == immeuble_id) - by_immeuble = [] - for row in session.execute(immeuble_stmt): - rev = row.total_revenus or 0.0 - reg = row.total_regles or 0.0 - taux = (reg / rev * 100) if rev > 0 else 100.0 - by_immeuble.append( - RevenuByImmeuble( - immeuble_id=row.id, - immeuble_code=row.code, - adresse=row.adresse, - ville=row.ville, - nb_lots=row.nb_lots or 0, - nb_locataires=row.nb_locataires or 0, - total_revenus=rev, - total_regles=reg, - total_impayes=row.total_impayes or 0.0, - taux_recouvrement=round(taux, 1), - ) - ) + by_immeuble = [_immeuble_response(row) for row in session.execute(immeuble_stmt)] # ========== Top impayes by locataire ========== impayes_stmt = ( @@ -471,43 +513,24 @@ async def get_immeubles_with_revenus( session: Session = Depends(get_session), ) -> list[RevenuByImmeuble]: """Retourne la liste des immeubles avec leurs stats de revenus.""" + # Tous les immeubles sont listes, y compris ceux sans aucun revenu. + montants = _revenus_par_immeuble() + effectifs = _effectifs_par_immeuble() stmt = ( select( Immeuble.id, Immeuble.code, Immeuble.adresse, Immeuble.ville, - func.count(func.distinct(Lot.id)).label("nb_lots"), - func.count(func.distinct(Locataire.id)).label("nb_locataires"), - func.sum(Revenu.total).label("total_revenus"), - func.sum(Revenu.regles).label("total_regles"), - func.sum(Revenu.impayes).label("total_impayes"), + effectifs.c.nb_lots, + effectifs.c.nb_locataires, + montants.c.total_revenus, + montants.c.total_regles, + montants.c.total_impayes, ) - .outerjoin(Lot, Lot.immeuble_id == Immeuble.id) - .outerjoin(Revenu, Revenu.lot_id == Lot.id) - .outerjoin(Locataire, Locataire.lot_id == Lot.id) - .group_by(Immeuble.id) + .outerjoin(montants, montants.c.immeuble_id == Immeuble.id) + .outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id) .order_by(Immeuble.code) ) - results = [] - for row in session.execute(stmt): - rev = row.total_revenus or 0.0 - reg = row.total_regles or 0.0 - taux = (reg / rev * 100) if rev > 0 else 100.0 - results.append( - RevenuByImmeuble( - immeuble_id=row.id, - immeuble_code=row.code, - adresse=row.adresse, - ville=row.ville, - nb_lots=row.nb_lots or 0, - nb_locataires=row.nb_locataires or 0, - total_revenus=rev, - total_regles=reg, - total_impayes=row.total_impayes or 0.0, - taux_recouvrement=round(taux, 1), - ) - ) - - return results + return [_immeuble_response(row) for row in session.execute(stmt)] diff --git a/tests/conftest.py b/tests/conftest.py index c089dc5..9b2da01 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -44,6 +44,27 @@ def db_session(tmp_path, monkeypatch): connection.reset_connection() +@pytest.fixture +def api_client(db_session): + """Client HTTP sur l'application FastAPI, branché sur la base de test. + + La dépendance `get_session` est surchargée pour partager la session du test : + les données créées dans le test sont visibles par les endpoints, sans passer + par la base réelle de l'utilisateur. + """ + from fastapi.testclient import TestClient + + from plesna_gerance.api.app import app + from plesna_gerance.database import get_session + + app.dependency_overrides[get_session] = lambda: db_session + try: + with TestClient(app) as client: + yield client + finally: + app.dependency_overrides.clear() + + @pytest.fixture def sample_data(): """Données extraites minimales mais complètes pour save_document.""" diff --git a/tests/test_revenus_agregats.py b/tests/test_revenus_agregats.py new file mode 100644 index 0000000..a095054 --- /dev/null +++ b/tests/test_revenus_agregats.py @@ -0,0 +1,99 @@ +"""Tests des agregats de revenus par immeuble. + +Ces endpoints joignent revenus et locataires a partir du lot. Comme un lot +accumule les locataires successifs, une jointure naive multiplie chaque ligne de +revenu par le nombre d'occupants passes et gonfle silencieusement les totaux : +aucune erreur, juste des montants faux qui derivent avec l'anciennete du parc. +""" + +import copy + +import pytest + +from plesna_gerance.database.models import Locataire, Revenu +from plesna_gerance.database.service import DatabaseService + + +@pytest.fixture +def lot_a_deux_locataires(db_session, sample_data): + """Un lot occupe successivement par deux locataires, un revenu chacun. + + Reproduit la situation courante d'une relocation : meme immeuble, meme lot, + deux noms differents sur deux comptes rendus. + """ + service = DatabaseService(db_session) + service.save_document(data=sample_data) + + suivant = copy.deepcopy(sample_data) + suivant["metadata"]["document"]["reference"] = "REF002" + suivant["metadata"]["document"]["date"] = "2024-02-15" + suivant["situation_locataires"][0]["locataire"]["nom"] = "MARTIN" + suivant["situation_locataires"][0]["lignes"][0]["loyers"] = 600.0 + suivant["situation_locataires"][0]["lignes"][0]["total"] = 600.0 + suivant["situation_locataires"][0]["lignes"][0]["regles"] = 400.0 + suivant["situation_locataires"][0]["lignes"][0]["impayes"] = 200.0 + service.save_document(data=suivant) + + # Le decor doit bien etre celui qu'on veut tester, sinon le test ne prouve rien. + assert db_session.query(Locataire).count() == 2 + assert db_session.query(Revenu).count() == 2 + return db_session + + +def _totaux_reels(session) -> tuple[float, float, float]: + revenus = session.query(Revenu).all() + return ( + sum(r.total for r in revenus), + sum(r.regles for r in revenus), + sum(r.impayes for r in revenus), + ) + + +def test_immeubles_ne_compte_pas_les_revenus_en_double( + api_client, lot_a_deux_locataires +): + """/api/revenus/immeubles somme chaque revenu une seule fois.""" + total, regles, impayes = _totaux_reels(lot_a_deux_locataires) + + response = api_client.get("/api/revenus/immeubles") + assert response.status_code == 200 + (immeuble,) = response.json() + + assert immeuble["total_revenus"] == total + assert immeuble["total_regles"] == regles + assert immeuble["total_impayes"] == impayes + + +def test_summary_ne_compte_pas_les_revenus_en_double( + api_client, lot_a_deux_locataires +): + """Le bloc by_immeuble de /api/revenus/summary somme comme les KPIs.""" + total, regles, impayes = _totaux_reels(lot_a_deux_locataires) + + response = api_client.get("/api/revenus/summary") + assert response.status_code == 200 + corps = response.json() + (immeuble,) = corps["by_immeuble"] + + assert immeuble["total_revenus"] == total + assert immeuble["total_regles"] == regles + assert immeuble["total_impayes"] == impayes + # Les KPIs globaux n'ont jamais eu le defaut : les deux doivent concorder. + assert corps["kpis"]["total_revenus"] == immeuble["total_revenus"] + + +def test_immeubles_compte_lots_et_locataires(api_client, lot_a_deux_locataires): + """Les effectifs restent justes : un lot, ses deux occupants successifs.""" + (immeuble,) = api_client.get("/api/revenus/immeubles").json() + + assert immeuble["nb_lots"] == 1 + assert immeuble["nb_locataires"] == 2 + + +def test_taux_recouvrement_reste_coherent(api_client, lot_a_deux_locataires): + """Le taux se deduit des totaux : il doit suivre la meme correction.""" + total, regles, _ = _totaux_reels(lot_a_deux_locataires) + + (immeuble,) = api_client.get("/api/revenus/immeubles").json() + + assert immeuble["taux_recouvrement"] == round(regles / total * 100, 1)