lafrite/pdf_oralia_vibe
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

fix: distingue les montants cumulables des soldes à date

Browse Source 
Chaque compte rendu reporte la dette du précédent dans une ligne
solde_anterieur. Les agrégats sommaient ces lignes comme le reste : la
même créance était recomptée à chaque document, et un remboursement ne
pouvait jamais s'inscrire — une dette soldée restait affichée à vie.

Sur la base réelle, la page Revenus annonçait ainsi 247 354 € d'impayés
pour une dette de 49 374 €, et désignait comme deuxième et troisième
débiteurs deux locataires à jour depuis avril (SURBECK 690,10 € et
GUINAIS 445,81 €, tous deux soldés).

Deux natures cohabitent, et c'est la colonne qui la porte, pas la ligne :
une ligne de report a un `total` déjà compté le mois d'avant, mais ses
`regles` sont un encaissement bien réel de la période. Écarter la ligne
entière ferait disparaître de l'argent reçu (1 298,81 € ici).

- flux (facturé, encaissé) : cumulés sur la période, report exclu ;
- stock (restant dû) : lu dans le dernier compte rendu de chaque immeuble ;
- taux de recouvrement : réglé sur facturé, report exclu des deux côtés,
  sans quoi rattraper une vieille dette ferait dépasser 100 %.

Résultat : 85 748 € facturés, 98,2 % de recouvrement, 49 374 € encore
dus. Les règles vivent dans services/revenus_query.py, pour que le
dashboard s'y branche au lieu de les réinventer.

/summary borne désormais tous ses blocs à la période demandée : les KPIs
et by_immeuble ignoraient `months` alors que by_month le respectait.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Bertrand Benjamin
2026-07-26 10:50:15 +02:00
parent 1c817a4ac9
commit 3e2d103929
4 changed files with 465 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files

262
src/plesna_gerance/api/routes/revenus.py
View File

@@ -15,6 +15,12 @@ from ...database.models import (
Lot,
Revenu,
)
from ...services.revenus_query import (
est_flux,
flux_par,
restant_du_par,
taux_de_recouvrement,
)
router = APIRouter(prefix="/api/revenus", tags=["revenus"])
@@ -132,27 +138,6 @@ class RevenusSummaryResponse(BaseModel):
# ============================================================
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 (
@@ -169,9 +154,6 @@ def _effectifs_par_immeuble():
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,
@@ -179,10 +161,39 @@ def _immeuble_response(row) -> RevenuByImmeuble:
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),
total_revenus=row.facture or 0.0,
total_regles=row.encaisse or 0.0,
total_impayes=row.restant_du or 0.0,
taux_recouvrement=taux_de_recouvrement(row.facture, row.facture_regle),
)
def _immeuble_stmt(date_debut: date | None = None):
"""Requete des immeubles avec leurs flux, leur restant du et leurs effectifs.
Les trois sous-requetes sont jointes plutot que calculees d'un bloc : chacune
a sa propre granularite (une ligne par revenu, par lot, par locataire) et les
melanger multiplierait les lignes entre elles.
"""
flux = flux_par(Document.immeuble_id, date_debut)
stock = restant_du_par(Document.immeuble_id, date_debut)
effectifs = _effectifs_par_immeuble()
return (
select(
Immeuble.id,
Immeuble.code,
Immeuble.adresse,
Immeuble.ville,
effectifs.c.nb_lots,
effectifs.c.nb_locataires,
flux.c.facture,
flux.c.encaisse,
flux.c.facture_regle,
stock.c.restant_du,
)
.outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id)
.outerjoin(flux, flux.c.cle == Immeuble.id)
.outerjoin(stock, stock.c.cle == Immeuble.id)
)
@@ -202,35 +213,32 @@ async def get_revenus_summary(
Inclut les KPIs, l'evolution mensuelle, la repartition par immeuble
et les locataires avec le plus d'impayes.
"""
# Base filters
filters = []
if immeuble_id:
filters.append(Lot.immeuble_id == immeuble_id)
# Calculate date range
today = date.today()
start_date = (today.replace(day=1) - timedelta(days=months * 31)).replace(day=1)
# ========== KPIs ==========
kpi_stmt = select(
func.sum(Revenu.total).label("total_revenus"),
func.sum(Revenu.loyers).label("total_loyers"),
func.sum(Revenu.taxes).label("total_taxes"),
func.sum(Revenu.provisions).label("total_provisions"),
func.sum(Revenu.regles).label("total_regles"),
func.sum(Revenu.impayes).label("total_impayes"),
).join(Lot, Revenu.lot_id == Lot.id)
# Les montants factures se cumulent sur la periode ; le restant du est lu
# dans le dernier compte rendu, sans quoi une meme dette serait recomptee a
# chaque document et un remboursement ne s'y verrait jamais.
flux = flux_par(Document.immeuble_id, start_date)
stock = restant_du_par(Document.immeuble_id, start_date)
if filters:
kpi_stmt = kpi_stmt.where(and_(*filters))
kpi_result = session.execute(kpi_stmt).first()
total_revenus = kpi_result.total_revenus or 0.0
total_regles = kpi_result.total_regles or 0.0
taux_recouvrement = (
(total_regles / total_revenus * 100) if total_revenus > 0 else 100.0
flux_stmt = select(
func.sum(flux.c.loyers).label("loyers"),
func.sum(flux.c.taxes).label("taxes"),
func.sum(flux.c.provisions).label("provisions"),
func.sum(flux.c.facture).label("facture"),
func.sum(flux.c.encaisse).label("encaisse"),
func.sum(flux.c.facture_regle).label("facture_regle"),
)
stock_stmt = select(func.sum(stock.c.restant_du))
if immeuble_id:
flux_stmt = flux_stmt.where(flux.c.cle == immeuble_id)
stock_stmt = stock_stmt.where(stock.c.cle == immeuble_id)
totaux = session.execute(flux_stmt).first()
restant_du = session.execute(stock_stmt).scalar() or 0.0
# Count active locataires and occupied lots
locataires_stmt = (
@@ -253,41 +261,43 @@ async def get_revenus_summary(
nb_lots = session.execute(lots_stmt).scalar() or 0
kpis = RevenuKpiResponse(
total_revenus=total_revenus,
total_loyers=kpi_result.total_loyers or 0.0,
total_taxes=kpi_result.total_taxes or 0.0,
total_provisions=kpi_result.total_provisions or 0.0,
total_regles=total_regles,
total_impayes=kpi_result.total_impayes or 0.0,
taux_recouvrement=round(taux_recouvrement, 1),
total_revenus=totaux.facture or 0.0,
total_loyers=totaux.loyers or 0.0,
total_taxes=totaux.taxes or 0.0,
total_provisions=totaux.provisions or 0.0,
total_regles=totaux.encaisse or 0.0,
total_impayes=restant_du,
taux_recouvrement=taux_de_recouvrement(totaux.facture, totaux.facture_regle),
nb_locataires_actifs=nb_locataires,
nb_lots_occupes=nb_lots,
)
# ========== Monthly evolution ==========
# Les montants du mois sont des flux, report exclu, pour que deux mois
# soient comparables. L'impaye, lui, reste le solde constate ce mois-la :
# la courbe montre l'evolution de la dette, pas son accumulation.
mois = func.strftime("%Y-%m", Document.date)
flux_mois = est_flux()
monthly_stmt = (
select(
func.strftime("%Y-%m", Document.date).label("month"),
func.sum(Revenu.loyers).label("loyers"),
func.sum(Revenu.taxes).label("taxes"),
func.sum(Revenu.provisions).label("provisions"),
func.sum(Revenu.total).label("total"),
mois.label("month"),
func.sum(Revenu.loyers).filter(flux_mois).label("loyers"),
func.sum(Revenu.taxes).filter(flux_mois).label("taxes"),
func.sum(Revenu.provisions).filter(flux_mois).label("provisions"),
func.sum(Revenu.total).filter(flux_mois).label("total"),
func.sum(Revenu.regles).label("regles"),
func.sum(Revenu.impayes).label("impayes"),
)
.join(Document, Revenu.document_id == Document.id)
.join(Lot, Revenu.lot_id == Lot.id)
.where(Document.date >= start_date)
.group_by(func.strftime("%Y-%m", Document.date))
.order_by(func.strftime("%Y-%m", Document.date))
.group_by(mois)
.order_by(mois)
)
if immeuble_id:
monthly_stmt = monthly_stmt.where(Lot.immeuble_id == immeuble_id)
monthly_stmt = monthly_stmt.where(Document.immeuble_id == immeuble_id)
monthly_data = []
for row in session.execute(monthly_stmt):
monthly_data.append(
monthly_data = [
RevenuMonthlyPoint(
month=row.month,
loyers=row.loyers or 0.0,
@@ -297,35 +307,27 @@ async def get_revenus_summary(
regles=row.regles or 0.0,
impayes=row.impayes or 0.0,
)
)
for row in session.execute(monthly_stmt)
]
# ========== 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,
effectifs.c.nb_lots,
effectifs.c.nb_locataires,
montants.c.total_revenus,
montants.c.total_regles,
montants.c.total_impayes,
)
.join(montants, montants.c.immeuble_id == Immeuble.id)
.outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id)
.order_by(desc(montants.c.total_revenus))
)
# Seuls les immeubles ayant produit des revenus sur la periode y figurent.
immeuble_stmt = _immeuble_stmt(start_date).order_by(desc("facture"))
if immeuble_id:
immeuble_stmt = immeuble_stmt.where(Immeuble.id == immeuble_id)
by_immeuble = [_immeuble_response(row) for row in session.execute(immeuble_stmt)]
by_immeuble = [
_immeuble_response(row)
for row in session.execute(immeuble_stmt)
if row.facture is not None
]
# ========== Top impayes by locataire ==========
# Le classement porte sur la dette encore due au dernier compte rendu. Sur
# un cumul, un locataire ayant solde son retard resterait affiche comme
# debiteur indefiniment : le remboursement ne s'y inscrivait jamais.
dette = restant_du_par(Revenu.locataire_id, start_date)
flux_locataire = flux_par(Revenu.locataire_id, start_date)
impayes_stmt = (
select(
Locataire.id,
@@ -334,26 +336,27 @@ async def get_revenus_summary(
Lot.numero,
Immeuble.code,
Immeuble.adresse,
func.sum(Revenu.total).label("total_revenus"),
func.sum(Revenu.regles).label("total_regles"),
func.sum(Revenu.impayes).label("total_impayes"),
func.count(Revenu.id).label("nb_mois"),
flux_locataire.c.facture,
flux_locataire.c.encaisse,
dette.c.restant_du,
func.count(func.distinct(Revenu.document_id)).label("nb_mois"),
)
.join(Lot, Revenu.lot_id == Lot.id)
.join(Locataire, Revenu.locataire_id == Locataire.id)
.select_from(Locataire)
.join(dette, dette.c.cle == Locataire.id)
.join(Lot, Locataire.lot_id == Lot.id)
.join(Immeuble, Lot.immeuble_id == Immeuble.id)
.outerjoin(flux_locataire, flux_locataire.c.cle == Locataire.id)
.outerjoin(Revenu, Revenu.locataire_id == Locataire.id)
.where(dette.c.restant_du > 0)
.group_by(Locataire.id)
.having(func.sum(Revenu.impayes) > 0)
.order_by(desc("total_impayes"))
.order_by(desc("restant_du"))
.limit(10)
)
if immeuble_id:
impayes_stmt = impayes_stmt.where(Lot.immeuble_id == immeuble_id)
top_impayes = []
for row in session.execute(impayes_stmt):
top_impayes.append(
top_impayes = [
RevenuByLocataire(
locataire_id=row.id,
locataire_nom=row.nom,
@@ -361,12 +364,13 @@ async def get_revenus_summary(
immeuble_code=row.code,
immeuble_adresse=row.adresse,
date_debut=str(row.date_debut) if row.date_debut else None,
total_revenus=row.total_revenus or 0.0,
total_regles=row.total_regles or 0.0,
total_impayes=row.total_impayes or 0.0,
total_revenus=row.facture or 0.0,
total_regles=row.encaisse or 0.0,
total_impayes=row.restant_du or 0.0,
nb_mois=row.nb_mois or 0,
)
)
for row in session.execute(impayes_stmt)
]
return RevenusSummaryResponse(
kpis=kpis,
@@ -383,6 +387,8 @@ async def get_revenus_by_lot(
session: Session = Depends(get_session),
) -> list[RevenuByLot]:
"""Retourne les revenus agreges par lot."""
flux = flux_par(Revenu.lot_id)
dette = restant_du_par(Revenu.lot_id)
stmt = (
select(
Lot.id,
@@ -390,9 +396,9 @@ async def get_revenus_by_lot(
Lot.type,
Immeuble.code,
func.max(Locataire.nom).label("locataire_nom"),
func.sum(Revenu.total).label("total_revenus"),
func.sum(Revenu.regles).label("total_regles"),
func.sum(Revenu.impayes).label("total_impayes"),
flux.c.facture,
flux.c.encaisse,
dette.c.restant_du,
func.max(Document.date).label("derniere_date"),
)
.join(Immeuble, Lot.immeuble_id == Immeuble.id)
@@ -402,31 +408,30 @@ async def get_revenus_by_lot(
Locataire,
and_(Locataire.lot_id == Lot.id, Locataire.date_fin.is_(None)),
)
.outerjoin(flux, flux.c.cle == Lot.id)
.outerjoin(dette, dette.c.cle == Lot.id)
.group_by(Lot.id)
.order_by(desc("total_impayes"), desc("total_revenus"))
.order_by(desc("restant_du"), desc("facture"))
.limit(limit)
)
if immeuble_id:
stmt = stmt.where(Lot.immeuble_id == immeuble_id)
results = []
for row in session.execute(stmt):
results.append(
return [
RevenuByLot(
lot_id=row.id,
lot_numero=row.numero,
lot_type=row.type,
immeuble_code=row.code,
locataire_nom=row.locataire_nom,
total_revenus=row.total_revenus or 0.0,
total_regles=row.total_regles or 0.0,
total_impayes=row.total_impayes or 0.0,
total_revenus=row.facture or 0.0,
total_regles=row.encaisse or 0.0,
total_impayes=row.restant_du or 0.0,
derniere_date=str(row.derniere_date) if row.derniere_date else None,
)
)
return results
for row in session.execute(stmt)
]
@router.get("/details", response_model=list[RevenuDetailResponse])
@@ -514,23 +519,6 @@ async def get_immeubles_with_revenus(
) -> 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,
effectifs.c.nb_lots,
effectifs.c.nb_locataires,
montants.c.total_revenus,
montants.c.total_regles,
montants.c.total_impayes,
)
.outerjoin(montants, montants.c.immeuble_id == Immeuble.id)
.outerjoin(effectifs, effectifs.c.immeuble_id == Immeuble.id)
.order_by(Immeuble.code)
)
stmt = _immeuble_stmt().order_by(Immeuble.code)
return [_immeuble_response(row) for row in session.execute(stmt)]

138
src/plesna_gerance/services/revenus_query.py Normal file
View File

@@ -0,0 +1,138 @@
"""Règles d'agrégation des revenus : ce qui se cumule et ce qui ne se cumule pas.
Deux natures de grandeurs cohabitent dans la table ``revenus``, et les
confondre fausse tous les totaux :
- un **flux** est un événement daté (un loyer facturé en mars, un paiement reçu
en avril). Il s'additionne dans le temps et entre les lots ;
- un **stock** est une photo à un instant (ce qui reste dû au 22/06). Il
s'additionne entre les lots à une même date, jamais dans le temps :
additionner deux photos du même solde compte deux fois la même dette.
Chaque compte rendu reporte la dette du précédent dans une ligne
``solde_anterieur``. Cumuler ces lignes sur une période revient donc à recompter
la même créance autant de fois qu'il y a de documents — et rend surtout
impossible d'enregistrer un remboursement : une dette soldée resterait dans le
total à vie.
Attention, **c'est la colonne qui porte la nature, pas la ligne**. Une ligne de
report contient les deux : sa colonne ``total`` est un stock déjà compté le mois
précédent, mais sa colonne ``regles`` est un encaissement bien réel de la
période. Écarter la ligne entière ferait disparaître de l'argent reçu.
"""
from datetime import date
from sqlalchemy import and_, func, select
from ..database.models import Document, Revenu
#: Type des lignes qui reportent le solde du compte rendu précédent.
TYPE_LIGNE_REPORT = "solde_anterieur"
def est_flux():
"""Condition : la ligne décrit un événement de la période, pas un report."""
return Revenu.type_ligne != TYPE_LIGNE_REPORT
def _borner(stmt, date_debut: date | None, date_fin: date | None):
if date_debut is not None:
stmt = stmt.where(Document.date >= date_debut)
if date_fin is not None:
stmt = stmt.where(Document.date <= date_fin)
return stmt
def derniers_comptes_rendus(
date_debut: date | None = None, date_fin: date | None = None
):
"""Sous-requête : date du dernier compte rendu de chaque immeuble.
Le dernier compte rendu est retenu **par immeuble** : avec plusieurs
immeubles, le dernier document tous immeubles confondus n'en décrirait
qu'un seul et les autres perdraient leur solde.
"""
stmt = select(
Document.immeuble_id.label("immeuble_id"),
func.max(Document.date).label("date"),
).group_by(Document.immeuble_id)
return _borner(stmt, date_debut, date_fin).subquery()
def restant_du_par(
cle, date_debut: date | None = None, date_fin: date | None = None
):
"""Sous-requête : restant dû (stock) regroupé par `cle`.
Seules les lignes du dernier compte rendu de chaque immeuble sont lues :
c'est la seule photo à jour. Un lot absent de ce compte rendu est sorti de
la gestion et ne compte plus — les présences observées sont contiguës, une
absence n'est jamais un simple trou.
Args:
cle: colonne de regroupement (``Revenu.lot_id``, ``Document.immeuble_id``…)
date_debut: borne basse optionnelle sur la date du document
date_fin: borne haute optionnelle
Returns:
Sous-requête exposant `cle` et ``restant_du``.
"""
derniers = derniers_comptes_rendus(date_debut, date_fin)
return (
select(
cle.label("cle"),
func.sum(Revenu.impayes).label("restant_du"),
)
.join(Document, Revenu.document_id == Document.id)
.join(
derniers,
and_(
derniers.c.immeuble_id == Document.immeuble_id,
derniers.c.date == Document.date,
),
)
.group_by(cle)
.subquery()
)
def flux_par(cle, date_debut: date | None = None, date_fin: date | None = None):
"""Sous-requête : montants cumulables (flux) regroupés par `cle`.
Expose :
- ``facture`` : ce qui a été facturé sur la période, report exclu ;
- ``encaisse`` : tout ce qui a été reçu, **y compris** les règlements de
dettes anciennes portés par les lignes de report ;
- ``facture_regle`` : la part de ``facture`` qui a été réglée. Sert au taux
de recouvrement, qui doit comparer un périmètre homogène : rapporter
``encaisse`` à ``facture`` ferait dépasser 100 % dès qu'une vieille dette
est rattrapée.
"""
flux = est_flux()
stmt = (
select(
cle.label("cle"),
func.sum(Revenu.loyers).filter(flux).label("loyers"),
func.sum(Revenu.taxes).filter(flux).label("taxes"),
func.sum(Revenu.provisions).filter(flux).label("provisions"),
func.sum(Revenu.total).filter(flux).label("facture"),
func.sum(Revenu.regles).label("encaisse"),
func.sum(Revenu.regles).filter(flux).label("facture_regle"),
)
.join(Document, Revenu.document_id == Document.id)
.group_by(cle)
)
return _borner(stmt, date_debut, date_fin).subquery()
def taux_de_recouvrement(facture: float | None, facture_regle: float | None) -> float:
"""Part du facturé qui a été réglée, en pourcentage.
Sans rien de facturé, il n'y a rien à recouvrer : le taux vaut 100 %.
"""
facture = facture or 0.0
if facture <= 0:
return 100.0
return round((facture_regle or 0.0) / facture * 100, 1)

54
tests/test_revenus_agregats.py
View File

@@ -7,6 +7,7 @@ aucune erreur, juste des montants faux qui derivent avec l'anciennete du parc.
"""
import copy
from datetime import date, timedelta
import pytest
@@ -14,6 +15,19 @@ from plesna_gerance.database.models import Locataire, Revenu
from plesna_gerance.database.service import DatabaseService
def _mois_glissant(recul: int) -> str:
"""Date du 15 du mois, `recul` mois en arriere.
Les resumes ne portent que sur les derniers mois : des dates fixes
sortiraient de la fenetre avec le temps et videraient les tests de leur
substance sans jamais les faire echouer.
"""
jour = date.today().replace(day=15)
for _ in range(recul):
jour = (jour.replace(day=1) - timedelta(days=1)).replace(day=15)
return jour.isoformat()
@pytest.fixture
def lot_a_deux_locataires(db_session, sample_data):
"""Un lot occupe successivement par deux locataires, un revenu chacun.
@@ -22,11 +36,13 @@ def lot_a_deux_locataires(db_session, sample_data):
deux noms differents sur deux comptes rendus.
"""
service = DatabaseService(db_session)
service.save_document(data=sample_data)
premier = copy.deepcopy(sample_data)
premier["metadata"]["document"]["date"] = _mois_glissant(2)
service.save_document(data=premier)
suivant = copy.deepcopy(sample_data)
suivant["metadata"]["document"]["reference"] = "REF002"
suivant["metadata"]["document"]["date"] = "2024-02-15"
suivant["metadata"]["document"]["date"] = _mois_glissant(1)
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
@@ -40,46 +56,40 @@ def lot_a_deux_locataires(db_session, sample_data):
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),
)
#: Le decor : 500 puis 600 factures, 500 puis 400 regles, 200 encore dus.
FACTURE = 1100.0
ENCAISSE = 900.0
RESTANT_DU = 200.0
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
assert immeuble["total_revenus"] == FACTURE
assert immeuble["total_regles"] == ENCAISSE
assert immeuble["total_impayes"] == RESTANT_DU
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 immeuble["total_revenus"] == FACTURE
assert immeuble["total_regles"] == ENCAISSE
assert immeuble["total_impayes"] == RESTANT_DU
# Les deux blocs doivent raconter la meme chose.
assert corps["kpis"]["total_revenus"] == immeuble["total_revenus"]
assert corps["kpis"]["total_impayes"] == immeuble["total_impayes"]
def test_immeubles_compte_lots_et_locataires(api_client, lot_a_deux_locataires):
@@ -92,8 +102,6 @@ def test_immeubles_compte_lots_et_locataires(api_client, lot_a_deux_locataires):
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)
assert immeuble["taux_recouvrement"] == round(ENCAISSE / FACTURE * 100, 1)

148
tests/test_revenus_flux_stock.py Normal file
View File

@@ -0,0 +1,148 @@
"""Tests de la distinction flux / stock dans les agregats de revenus.
Chaque compte rendu reporte la dette du precedent dans une ligne
`solde_anterieur`. Cumuler ces lignes recompte la meme creance a chaque
document et, surtout, empeche un remboursement de s'inscrire : une dette soldee
resterait affichee a vie.
Le decor rejoue le cycle observe en production : un locataire laisse un impaye,
le compte rendu suivant le reporte, il le solde, puis le reporte disparait.
"""
import copy
from datetime import date, timedelta
import pytest
from plesna_gerance.database.service import DatabaseService
def _mois_glissant(recul: int) -> str:
jour = date.today().replace(day=15)
for _ in range(recul):
jour = (jour.replace(day=1) - timedelta(days=1)).replace(day=15)
return jour.isoformat()
def _ligne(type_ligne: str, **montants) -> dict:
base = {
"type": type_ligne,
"periode": {"debut": None, "fin": None},
"loyers": 0.0,
"taxes": 0.0,
"provisions": 0.0,
"total": 0.0,
"regles": 0.0,
"impayes": 0.0,
}
base.update(montants)
return base
def _compte_rendu(sample_data: dict, reference: str, recul: int, lignes: list) -> dict:
data = copy.deepcopy(sample_data)
data["metadata"]["document"]["reference"] = reference
data["metadata"]["document"]["date"] = _mois_glissant(recul)
data["situation_locataires"][0]["lignes"] = lignes
data["recapitulatif_operations"] = []
return data
@pytest.fixture
def dette_reportee_puis_soldee(db_session, sample_data):
"""Trois mois : un impaye nait, il est reporte, il est solde.
Mois 1 : loyer de 800 dont 300 impayes.
Mois 2 : les 300 sont reportes et regles ; loyer de 800 regle en entier.
Mois 3 : plus aucun report ; loyer de 800 regle en entier.
"""
service = DatabaseService(db_session)
service.save_document(
data=_compte_rendu(
sample_data,
"M1",
2,
[_ligne("loyer", loyers=800.0, total=800.0, regles=500.0, impayes=300.0)],
)
)
service.save_document(
data=_compte_rendu(
sample_data,
"M2",
1,
[
_ligne("solde_anterieur", loyers=300.0, total=300.0, regles=300.0),
_ligne("loyer", loyers=800.0, total=800.0, regles=800.0),
],
)
)
service.save_document(
data=_compte_rendu(
sample_data,
"M3",
0,
[_ligne("loyer", loyers=800.0, total=800.0, regles=800.0)],
)
)
return db_session
def test_le_restant_du_est_celui_du_dernier_compte_rendu(
api_client, dette_reportee_puis_soldee
):
"""La dette soldee disparait : le cumul afficherait encore 300."""
kpis = api_client.get("/api/revenus/summary").json()["kpis"]
assert kpis["total_impayes"] == 0.0
def test_le_facture_ignore_le_report(api_client, dette_reportee_puis_soldee):
"""3 loyers de 800 : le report de 300 n'est pas un revenu de plus."""
kpis = api_client.get("/api/revenus/summary").json()["kpis"]
assert kpis["total_revenus"] == 2400.0
def test_l_encaisse_retient_le_reglement_d_une_vieille_dette(
api_client, dette_reportee_puis_soldee
):
"""Les 300 regles sur la ligne de report sont de l'argent bien recu.
C'est ce qui interdit d'ecarter la ligne de report en bloc : sa colonne
`total` est un stock deja compte, mais sa colonne `regles` est un flux.
"""
kpis = api_client.get("/api/revenus/summary").json()["kpis"]
assert kpis["total_regles"] == 500.0 + 300.0 + 800.0 + 800.0
def test_le_taux_compare_un_perimetre_homogene(
api_client, dette_reportee_puis_soldee
):
"""Regle sur facture, report exclu des deux cotes : 2100 / 2400.
Rapporter l'encaisse (2400, rattrapage compris) au facture ferait afficher
un taux de 100 % alors qu'un impaye est ne sur la periode.
"""
kpis = api_client.get("/api/revenus/summary").json()["kpis"]
assert kpis["taux_recouvrement"] == round(2100.0 / 2400.0 * 100, 1)
def test_le_classement_des_impayes_oublie_qui_a_paye(
api_client, dette_reportee_puis_soldee
):
"""Un locataire a jour ne doit plus figurer parmi les debiteurs."""
top = api_client.get("/api/revenus/summary").json()["top_impayes"]
assert top == []
def test_les_mois_restent_comparables(api_client, dette_reportee_puis_soldee):
"""Chaque mois vaut son loyer : le report ne gonfle pas le mois 2."""
by_month = api_client.get("/api/revenus/summary").json()["by_month"]
assert [point["total"] for point in by_month] == [800.0, 800.0, 800.0]
# L'impaye reste le solde constate ce mois-la, pas un cumul.
assert [point["impayes"] for point in by_month] == [300.0, 0.0, 0.0]