""" tests/test_analysis.py ====================== Unit tests for tools/analysis.py and the /analysis API endpoint. NO LLM calls. NO real HTTP calls. NO filesystem access. All external dependencies are mocked. Run with : pytest tests/test_analysis.py -v """ import unittest import os import sys from unittest.mock import AsyncMock, patch import pytest sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from domain.tools.analysis import ( CENT, COEFS_ORIENTATION, COEFS_RENOVATION, DIST_EPSILON_KM, MIN_VENTES, PRIX_M2_MAX, PRIX_M2_MIN, TOP_N_VENTES, _ANNEE_CIBLE, _coef_terrain, _haversine, _normaliser_commune, get_surface_coefficient, ) # ══════════════════════════════════════════════════════════════════════════ # Pure function tests — no mocking needed # ══════════════════════════════════════════════════════════════════════════ class TestNormaliserCommune: def test_accents_supprimes(self): assert _normaliser_commune("Lézennes") == "LEZENNES" def test_tiret_remplace_par_espace(self): # Pure hyphen case, no ligatures assert _normaliser_commune("Saint-Andre-lez-Lille") == "SAINT ANDRE LEZ LILLE" def test_apostrophe_ascii_remplacee_par_espace(self): # U+0027 APOSTROPHE → space — use chr(39) to avoid smart-quote substitution in editors name = "Villeneuve-d" + chr(39) + "Ascq" assert _normaliser_commune(name) == "VILLENEUVE D ASCQ" def test_apostrophe_unicode_droppee(self): # U+2019 RIGHT SINGLE QUOTATION MARK is non-ASCII and dropped by encode("ascii","ignore") # Result: "dAscq" → no space inserted (the apostrophe is gone, not converted to space) assert _normaliser_commune("Villeneuve-d’Ascq") == "VILLENEUVE DASCQ" def test_ligature_oe_droppee(self): # œ (U+0153) is non-ASCII and dropped — documents current behavior, not a decomposition assert _normaliser_commune("Marcq-en-Barœul") == "MARCQ EN BARUL" def test_ligature_ae_droppee(self): # Æ (U+00C6) is non-ASCII and dropped assert _normaliser_commune("Æthelmere") == "THELMERE" def test_already_clean(self): assert _normaliser_commune("LILLE") == "LILLE" def test_minuscules_converties_en_majuscules(self): assert _normaliser_commune("wasquehal") == "WASQUEHAL" def test_double_espaces_fusionnes(self): result = _normaliser_commune("Saint André") assert " " not in result class TestCoefTerrain: def test_zero_terrain(self): assert _coef_terrain(0.0) == 1.00 def test_terrain_negatif(self): assert _coef_terrain(-50.0) == 1.00 def test_boundary_100(self): assert _coef_terrain(100.0) == 1.01 def test_boundary_101(self): assert _coef_terrain(101.0) == 1.03 def test_boundary_300(self): assert _coef_terrain(300.0) == 1.03 def test_boundary_301(self): assert _coef_terrain(301.0) == 1.05 def test_boundary_700(self): assert _coef_terrain(700.0) == 1.08 def test_boundary_1000(self): assert _coef_terrain(1000.0) == 1.11 def test_boundary_1001(self): assert _coef_terrain(1001.0) == 1.13 def test_boundary_2500(self): assert _coef_terrain(2500.0) == 1.20 def test_tres_grand_terrain(self): assert _coef_terrain(5000.0) == 1.25 class TestHaversine: def test_meme_point_distance_zero(self): assert _haversine(50.0, 3.0, 50.0, 3.0) == 0.0 def test_symetrie(self): d1 = _haversine(50.0, 3.0, 48.0, 2.0) d2 = _haversine(48.0, 2.0, 50.0, 3.0) assert abs(d1 - d2) < 1e-6 def test_lille_wasquehal_moins_de_5km(self): # Lille centre ≈ (50.630, 3.057), Wasquehal ≈ (50.672, 3.131) dist = _haversine(50.630, 3.057, 50.672, 3.131) assert dist < 10.0 def test_lille_paris_entre_200_et_250km(self): # Lille ≈ (50.63, 3.06), Paris ≈ (48.85, 2.35) dist = _haversine(50.63, 3.06, 48.85, 2.35) assert 200 < dist < 250 def test_distance_positive(self): assert _haversine(50.0, 3.0, 51.0, 4.0) > 0.0 class TestCoefsRenovation: def test_renovated_est_baseline(self): assert COEFS_RENOVATION["renovated"] == 1.00 def test_terrible_inferieur_a_un(self): assert COEFS_RENOVATION["terrible"] < 1.00 def test_premium_plus_superieur_a_un(self): assert COEFS_RENOVATION["premium plus"] > 1.00 def test_ordre_monotone_croissant(self): ordre = [ "terrible", "bad", "light renovation", "medium", "renovated", "good", "premium", "premium plus", ] valeurs = [COEFS_RENOVATION[k] for k in ordre] assert valeurs == sorted(valeurs) def test_tous_les_statuts_presents(self): from domain.tools.analysis import STATUTS_VALIDES assert set(COEFS_RENOVATION.keys()) == STATUTS_VALIDES class TestCoefsOrientation: def test_sud_est_maximum(self): assert COEFS_ORIENTATION["sud"] == max(COEFS_ORIENTATION.values()) def test_nord_est_minimum(self): assert COEFS_ORIENTATION["nord"] == min(COEFS_ORIENTATION.values()) def test_est_et_ouest_sont_baseline(self): assert COEFS_ORIENTATION["est"] == 1.00 assert COEFS_ORIENTATION["ouest"] == 1.00 def test_sud_superieur_nord(self): assert COEFS_ORIENTATION["sud"] > COEFS_ORIENTATION["nord"] # ══════════════════════════════════════════════════════════════════════════ # Evolution compounding — pure logic (no module state) # ══════════════════════════════════════════════════════════════════════════ class TestEvolutionCompounding: """ Tests the market-projection formula used in analyser_bien: prix_ajuste = prix_m2 × ∏(1 + evol_y / 100) for y = annee+1 … 2025 """ def _compound(self, annee_vente: int, evol_index: dict[tuple, float | None]) -> float: """Replicates the compounding logic from analyser_bien.""" compound = 1.0 for y in range(annee_vente + 1, _ANNEE_CIBLE + 1): evol = evol_index.get(("59290", y)) if evol is not None: compound *= 1.0 + evol / CENT return compound def test_vente_2025_aucun_ajustement(self): assert self._compound(2025, {}) == pytest.approx(1.0) def test_evolution_positive_augmente_le_prix(self): evol_index = {("59290", 2025): 5.0} assert self._compound(2024, evol_index) == pytest.approx(1.05) def test_evolution_negative_baisse_le_prix(self): """A market drop of 2% in 2025 should reduce a 2024 sale price by 2%.""" evol_index = {("59290", 2025): -2.0} assert self._compound(2024, evol_index) == pytest.approx(0.98) def test_deux_annees_composees(self): evol_index = {("59290", 2024): 3.0, ("59290", 2025): 2.0} expected = 1.03 * 1.02 assert self._compound(2023, evol_index) == pytest.approx(expected, rel=1e-9) def test_annee_manquante_traitee_comme_neutre(self): # Only 2025 is known; 2022, 2023, 2024 missing → treated as ×1.0 evol_index = {("59290", 2025): 4.0} assert self._compound(2021, evol_index) == pytest.approx(1.04) def test_compoundage_sur_cinq_ans(self): evol_index = { ("59290", 2022): 2.0, ("59290", 2023): -1.0, ("59290", 2024): 3.0, ("59290", 2025): 1.5, } expected = 1.02 * 0.99 * 1.03 * 1.015 assert self._compound(2021, evol_index) == pytest.approx(expected, rel=1e-9) def test_evolutions_toutes_negatives_produit_inferieur_a_un(self): evol_index = {("59290", 2024): -3.0, ("59290", 2025): -2.0} compound = self._compound(2023, evol_index) assert compound < 1.0 assert compound == pytest.approx(0.97 * 0.98, rel=1e-9) def test_annee_cible_borne_superieure(self): """Compounding never exceeds _ANNEE_CIBLE regardless of evol_index content.""" # An entry beyond _ANNEE_CIBLE must not be picked up evol_index = {("59290", _ANNEE_CIBLE + 1): 99.0} assert self._compound(_ANNEE_CIBLE - 1, evol_index) == pytest.approx(1.0) class TestRecencyWeights: def test_ratio_2025_sur_2021(self): from domain.tools.analysis import _RECENCY_WEIGHTS assert _RECENCY_WEIGHTS[2025] / _RECENCY_WEIGHTS[2021] == pytest.approx(16 / 3) def test_poids_strictement_croissants(self): from domain.tools.analysis import _RECENCY_WEIGHTS annees = sorted(_RECENCY_WEIGHTS) valeurs = [_RECENCY_WEIGHTS[a] for a in annees] assert valeurs == sorted(valeurs) def test_vente_recente_domine_vente_ancienne_eloignee(self): """A 2025 sale 300m away must outweigh a 2021 sale 500m away.""" from domain.tools.analysis import _RECENCY_WEIGHTS w_2021_eloigne = (1.0 / (0.50 + DIST_EPSILON_KM)) * _RECENCY_WEIGHTS[2021] w_2025_proche = (1.0 / (0.30 + DIST_EPSILON_KM)) * _RECENCY_WEIGHTS[2025] assert w_2025_proche > w_2021_eloigne class TestOutlierFilter: def test_prix_m2_normal_inclus(self): assert PRIX_M2_MIN <= 3000.0 <= PRIX_M2_MAX def test_prix_m2_trop_bas_exclu(self): prix_m2 = 200.0 assert not (PRIX_M2_MIN <= prix_m2 <= PRIX_M2_MAX) def test_prix_m2_trop_haut_exclu(self): prix_m2 = 20_000.0 assert not (PRIX_M2_MIN <= prix_m2 <= PRIX_M2_MAX) def test_bornes_incluses(self): assert PRIX_M2_MIN <= PRIX_M2_MIN <= PRIX_M2_MAX assert PRIX_M2_MIN <= PRIX_M2_MAX <= PRIX_M2_MAX async def test_transaction_hors_limites_non_chargee(self, tmp_path): """_load_transactions skips rows where prix_m2 is outside [500, 15000].""" import csv as csv_mod import domain.tools.analysis as ana csv_file = tmp_path / "59.csv" fieldnames = [ "nature_mutation", "type_local", "surface_reelle_bati", "valeur_fonciere", "id_mutation", "latitude", "longitude", "nom_commune", "code_postal", "date_mutation", "adresse_nom_voie", "adresse_numero", ] rows = [ # valid row: 3000 €/m² {"nature_mutation": "Vente", "type_local": "Appartement", "surface_reelle_bati": "50", "valeur_fonciere": "150000", "id_mutation": "M1", "latitude": "50.67", "longitude": "3.13", "nom_commune": "Wasquehal", "code_postal": "59290", "date_mutation": "2024-01-01", "adresse_nom_voie": "RUE TEST", "adresse_numero": "1"}, # outlier row: 50 €/m² {"nature_mutation": "Vente", "type_local": "Appartement", "surface_reelle_bati": "50", "valeur_fonciere": "2500", "id_mutation": "M2", "latitude": "50.67", "longitude": "3.13", "nom_commune": "Wasquehal", "code_postal": "59290", "date_mutation": "2024-01-02", "adresse_nom_voie": "RUE TEST", "adresse_numero": "2"}, # outlier row: 50 000 €/m² {"nature_mutation": "Vente", "type_local": "Appartement", "surface_reelle_bati": "50", "valeur_fonciere": "2500000", "id_mutation": "M3", "latitude": "50.67", "longitude": "3.13", "nom_commune": "Wasquehal", "code_postal": "59290", "date_mutation": "2024-01-03", "adresse_nom_voie": "RUE TEST", "adresse_numero": "3"}, ] with open(csv_file, "w", newline="", encoding="utf-8") as f: writer = csv_mod.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) original_root = ana._ROOT original_annees = ana._ANNEES_DVF try: ana._ROOT = str(tmp_path.parent) ana._ANNEES_DVF = [2024] # Place CSV at expected path data_dir = tmp_path.parent / "data" / "2024" data_dir.mkdir(parents=True, exist_ok=True) import shutil shutil.copy(csv_file, data_dir / "59.csv") loaded = ana._load_transactions() finally: ana._ROOT = original_root ana._ANNEES_DVF = original_annees assert len(loaded) == 1 assert loaded[0]["prix_m2"] == pytest.approx(3000.0) # ══════════════════════════════════════════════════════════════════════════ # analyser_bien() — mocked unit tests # ══════════════════════════════════════════════════════════════════════════ # --- Fixtures de données --- _SAMPLE_TRANSACTIONS = [ { "annee": 2024, "date": "2024-05-10", "commune": "Wasquehal", "code_postal": "59290", "voie": "RUE DE LA PAIX", "no_voie": "12", "valeur": 200000.0, "surface": 65.0, "type_local": "Appartement", "max_surface": 65.0, "lat": 50.671, "lon": 3.131, "prix_m2": 3076.92, }, { "annee": 2025, "date": "2025-02-01", "commune": "Wasquehal", "code_postal": "59290", "voie": "AVENUE DES FLEURS", "no_voie": "5", "valeur": 210000.0, "surface": 68.0, "type_local": "Appartement", "max_surface": 68.0, "lat": 50.675, "lon": 3.135, "prix_m2": 3088.24, }, { "annee": 2023, "date": "2023-11-15", "commune": "Lille", "code_postal": "59000", "voie": "RUE NATIONALE", "no_voie": "41", "valeur": 195000.0, "surface": 55.0, "type_local": "Appartement", "max_surface": 55.0, "lat": 50.630, "lon": 3.057, "prix_m2": 3545.45, }, { "annee": 2022, "date": "2022-07-20", "commune": "Wasquehal", "code_postal": "59290", "voie": "RUE GAMBETTA", "no_voie": "8", "valeur": 180000.0, "surface": 60.0, "type_local": "Appartement", "max_surface": 60.0, "lat": 50.673, "lon": 3.132, "prix_m2": 3000.0, }, { "annee": 2021, "date": "2021-03-10", "commune": "Wasquehal", "code_postal": "59290", "voie": "BOULEVARD DE LA LIBERTE", "no_voie": "2", "valeur": 170000.0, "surface": 58.0, "type_local": "Maison", "max_surface": 58.0, "lat": 50.670, "lon": 3.130, "prix_m2": 2931.03, }, { "annee": 2021, "date": "2021-09-05", "commune": "Wasquehal", "code_postal": "59290", "voie": "RUE DU MOULIN", "no_voie": "3", "valeur": 165000.0, "surface": 55.0, "type_local": "Appartement", "max_surface": 55.0, "lat": 50.672, "lon": 3.133, "prix_m2": 3000.0, }, ] _SAMPLE_PRIX_EVOLUTION = [ {"code_postal": "59290", "annee": 2022, "type_local": "Appartement", "evolution_m2_pct": 2.0}, {"code_postal": "59290", "annee": 2023, "type_local": "Appartement", "evolution_m2_pct": 1.0}, {"code_postal": "59290", "annee": 2024, "type_local": "Appartement", "evolution_m2_pct": 3.0}, {"code_postal": "59290", "annee": 2025, "type_local": "Appartement", "evolution_m2_pct": 1.5}, {"code_postal": "59000", "annee": 2024, "type_local": "Appartement", "evolution_m2_pct": -1.0}, {"code_postal": "59000", "annee": 2025, "type_local": "Appartement", "evolution_m2_pct": 0.5}, {"code_postal": "59290", "annee": 2022, "type_local": "Maison", "evolution_m2_pct": 1.5}, {"code_postal": "59290", "annee": 2025, "type_local": "Maison", "evolution_m2_pct": -2.0}, ] _SAMPLE_COORDS = { "WASQUEHAL": (50.672, 3.131), "LILLE": (50.630, 3.057), } @pytest.fixture def patched_analysis(monkeypatch): """Patch module globals so analyser_bien uses test data instead of real files/DB.""" import domain.tools.analysis as ana monkeypatch.setattr(ana, "_transactions", list(_SAMPLE_TRANSACTIONS)) monkeypatch.setattr(ana, "_prix_evolution", list(_SAMPLE_PRIX_EVOLUTION)) monkeypatch.setattr(ana, "_commune_coords", dict(_SAMPLE_COORDS)) return ana class TestAnalyserBien: async def test_retourne_les_cles_attendues(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "sud", ) assert "estimation" in result assert "transactions_reference" in result assert "google_maps_url" in result async def test_prix_m2_marche_positif(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) assert result["estimation"]["prix_m2_marche"] > 0 assert result["estimation"]["prix_total_estime"] > 0 async def test_surface_double_double_le_prix_total(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) r50 = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 50.0, "renovated", "appartement", "est", ) r100 = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 100.0, "renovated", "appartement", "est", ) ratio = r100["estimation"]["prix_total_estime"] / r50["estimation"]["prix_total_estime"] assert ratio == pytest.approx(2.0, rel=0.01) async def test_orientation_sud_superieure_a_nord(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) r_sud = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "sud", ) r_nord = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "nord", ) assert r_sud["estimation"]["prix_m2_estime"] > r_nord["estimation"]["prix_m2_estime"] async def test_statut_premium_superieur_a_terrible(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) r_premium = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "premium plus", "appartement", "est", ) r_terrible = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "terrible", "appartement", "est", ) assert r_premium["estimation"]["prix_m2_estime"] > r_terrible["estimation"]["prix_m2_estime"] async def test_terrain_augmente_le_prix(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) r_sans = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", 0.0, ) r_avec = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", 500.0, ) assert r_avec["estimation"]["prix_m2_estime"] > r_sans["estimation"]["prix_m2_estime"] async def test_max_20_transactions_reference(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) assert len(result["transactions_reference"]) <= TOP_N_VENTES async def test_filtre_type_bien_maison(self, patched_analysis, monkeypatch): """Only the one Maison transaction should be in the pool — but < 5 → ValueError.""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) with pytest.raises(ValueError, match="insuffisantes"): await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 100.0, "renovated", "maison", "est", ) async def test_transactions_vides_leve_runtime_error(self, monkeypatch): import domain.tools.analysis as ana monkeypatch.setattr(ana, "_transactions", []) with pytest.raises(RuntimeError): await ana.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) async def test_vente_2025_evol_marche_zero(self, patched_analysis, monkeypatch): """A 2025 sale needs no forward compounding — evol_marche_pct must be 0%.""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) ref_2025 = next( (t for t in result["transactions_reference"] if t["annee"] == 2025), None ) assert ref_2025 is not None assert ref_2025["evol_marche_pct"] == pytest.approx(0.0, abs=0.01) assert ref_2025["prix_m2_ajuste"] == pytest.approx(ref_2025["prix_m2_vente"], abs=1.0) async def test_vente_2024_ajustee_par_evol_2025(self, patched_analysis, monkeypatch): """A 2024 sale from CP 59290 must be adjusted by +1.5% (evol_2025 = 1.5).""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) ref_2024 = next( (t for t in result["transactions_reference"] if t["annee"] == 2024 and t["code_postal"] == "59290"), None ) assert ref_2024 is not None expected_ajuste = ref_2024["prix_m2_vente"] * 1.015 assert ref_2024["prix_m2_ajuste"] == pytest.approx(expected_ajuste, rel=0.01) assert ref_2024["evol_marche_pct"] == pytest.approx(1.5, abs=0.01) async def test_transaction_reference_contient_les_champs_requis(self, patched_analysis, monkeypatch): monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) for ref in result["transactions_reference"]: for key in ("commune", "code_postal", "annee", "date", "surface_m2", "prix_total", "prix_m2_vente", "evol_marche_pct", "prix_m2_ajuste", "distance_km", "google_maps_url"): assert key in ref, f"Clé manquante dans transactions_reference : {key!r}" # ══════════════════════════════════════════════════════════════════════════ # /analysis API endpoint tests — mocked lifespan + analyser_bien # ══════════════════════════════════════════════════════════════════════════ _ANALYSIS_RESULT = { "adresse": "12 rue de la paix", "code_postal": 59290, "ville": "wasquehal", "surface_m2": 65.0, "type_bien": "appartement", "statut": "renovated", "orientation": "sud", "taille_terrain": 0.0, "google_maps_url": "https://www.google.com/maps/search/?api=1&query=12+rue+de+la+paix", "estimation": { "prix_m2_marche": 3200.0, "coef_renovation": 1.0, "coef_orientation": 1.1, "coef_terrain": 1.0, "coef_total": 1.1, "prix_m2_estime": 3520.0, "prix_total_estime": 228800, }, "transactions_reference": [], } _VALID_PAYLOAD = { "adresse": "12 rue de la paix", "code_postal": 59290, "ville": "wasquehal", "surface_m2": 65.0, "statut": "renovated", "type_bien": "appartement", "orientation": "sud", "taille_terrain": 0.0, } @pytest.fixture async def api_client(): """ Async FastAPI test client with all startup side-effects mocked out: - setup_db → no-op - init_analysis → async no-op - analyser_bien → returns _ANALYSIS_RESULT API key check is disabled because API_KEY env var is not set in tests. """ from httpx import ASGITransport, AsyncClient import api as api_module with ( patch("api.setup_db"), patch("api.init_analysis", new_callable=AsyncMock), patch("domain.services.analysis_service.analyser_bien", new_callable=AsyncMock, return_value=_ANALYSIS_RESULT), ): async with AsyncClient( transport=ASGITransport(app=api_module.app), base_url="http://test", ) as client: yield client class TestAnalysisEndpoint: async def test_requete_valide_retourne_200(self, api_client): r = await api_client.post("/analysis", json=_VALID_PAYLOAD) assert r.status_code == 200 async def test_reponse_contient_estimation(self, api_client): r = await api_client.post("/analysis", json=_VALID_PAYLOAD) body = r.json() assert "estimation" in body assert body["estimation"]["prix_total_estime"] == 228800 async def test_reponse_contient_transactions_reference(self, api_client): r = await api_client.post("/analysis", json=_VALID_PAYLOAD) body = r.json() assert "transactions_reference" in body assert isinstance(body["transactions_reference"], list) async def test_champ_surface_manquant_retourne_422(self, api_client): payload = {k: v for k, v in _VALID_PAYLOAD.items() if k != "surface_m2"} r = await api_client.post("/analysis", json=payload) assert r.status_code == 422 async def test_statut_invalide_retourne_422(self, api_client): r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "statut": "parfait"}) assert r.status_code == 422 async def test_type_bien_invalide_retourne_422(self, api_client): r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "type_bien": "studio"}) assert r.status_code == 422 async def test_orientation_invalide_retourne_422(self, api_client): r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "orientation": "est-nord-est"}) assert r.status_code == 422 async def test_surface_negative_retourne_422(self, api_client): r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "surface_m2": -10.0}) assert r.status_code == 422 async def test_valueerror_retourne_400(self): """analyser_bien raising ValueError → 400 (adresse introuvable, données insuffisantes…).""" from httpx import ASGITransport, AsyncClient import api as api_module with ( patch("api.setup_db"), patch("api.init_analysis", new_callable=AsyncMock), patch("domain.services.analysis_service.analyser_bien", new_callable=AsyncMock, side_effect=ValueError("Données insuffisantes")), ): async with AsyncClient( transport=ASGITransport(app=api_module.app), base_url="http://test", ) as client: r = await client.post("/analysis", json=_VALID_PAYLOAD) assert r.status_code == 400 async def test_runtimeerror_retourne_503(self): """analyser_bien raising RuntimeError → 503 (données DVF introuvables).""" from httpx import ASGITransport, AsyncClient import api as api_module with ( patch("api.setup_db"), patch("api.init_analysis", new_callable=AsyncMock), patch("domain.services.analysis_service.analyser_bien", new_callable=AsyncMock, side_effect=RuntimeError("Données DVF indisponibles")), ): async with AsyncClient( transport=ASGITransport(app=api_module.app), base_url="http://test", ) as client: r = await client.post("/analysis", json=_VALID_PAYLOAD) assert r.status_code == 503 async def test_payload_vide_retourne_422(self, api_client): r = await api_client.post("/analysis", json={}) assert r.status_code == 422 async def test_tous_les_statuts_valides_acceptes(self, api_client): statuts = [ "terrible", "bad", "light renovation", "medium", "renovated", "good", "premium", "premium plus", ] for statut in statuts: r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "statut": statut}) assert r.status_code == 200, f"Statut {statut!r} refusé — attendu 200, obtenu {r.status_code}" async def test_toutes_les_orientations_valides_acceptees(self, api_client): orientations = ["nord", "nord-est", "est", "sud-est", "sud", "sud-ouest", "ouest", "nord-ouest"] for orientation in orientations: r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "orientation": orientation}) assert r.status_code == 200, f"Orientation {orientation!r} refusée — attendu 200" async def test_maison_et_appartement_acceptes(self, api_client): for type_bien in ("appartement", "maison"): r = await api_client.post("/analysis", json={**_VALID_PAYLOAD, "type_bien": type_bien}) assert r.status_code == 200, f"type_bien {type_bien!r} refusé — attendu 200" class TestSurfaceCoefficient(unittest.TestCase): def test_coefficients(self): # Format: (input_surface, expected_output) # Values derived from the formula: 1.0 + 0.75 * log10(90/surface) test_cases = [ (15, 1.600), # Clamped at Max (20, 1.490), (21, 1.474), (30, 1.358), (50, 1.191), (60, 1.132), # Start of comfort zone (65, 1.106), (90, 1.000), # Neutral anchor (100, 0.966), (110, 0.935), (120, 0.906), # End of comfort zone (130, 0.880), (135, 0.868), (145, 0.845), (155, 0.823), (175, 0.783), (195, 0.748), (200, 0.740), (250, 0.700), # Clamped at Min ] for surface, expected in test_cases: with self.subTest(surface=surface): result = get_surface_coefficient(surface) self.assertEqual(result, expected, f"Failed for surface: {surface}") def test_extreme_values(self): """Ensure safety with edge cases""" self.assertEqual(get_surface_coefficient(0), 1.6) self.assertEqual(get_surface_coefficient(-10), 1.6) self.assertEqual(get_surface_coefficient(10000), 0.7) # ══════════════════════════════════════════════════════════════════════════ # Limit / boundary case tests # ══════════════════════════════════════════════════════════════════════════ class TestCoefTerrainLimites: def test_valeur_positive_infime(self): """Un terrain de 0.001 m² est > 0 → premier palier (1.01).""" assert _coef_terrain(0.001) == 1.01 def test_juste_en_dessous_100(self): assert _coef_terrain(99.9) == 1.01 def test_juste_au_dessus_100(self): assert _coef_terrain(100.001) == 1.03 def test_exactement_500(self): assert _coef_terrain(500.0) == 1.05 def test_juste_au_dessus_500(self): assert _coef_terrain(500.001) == 1.08 def test_exactement_2500(self): assert _coef_terrain(2500.0) == 1.20 def test_juste_au_dessus_2500(self): assert _coef_terrain(2500.001) == 1.25 class TestSurfaceCoefficientLimites(unittest.TestCase): def test_juste_en_dessous_15_clampe_max(self): """14.9 < 15 → également capped à 1.6.""" self.assertEqual(get_surface_coefficient(14.9), 1.6) def test_juste_au_dessus_15_utilise_formule(self): """16 > 15 → formule utilisée, plus capped au maximum.""" # 1.0 + 0.75 * log10(90/16) = 1.5626 → arrondi à 1.563 self.assertEqual(get_surface_coefficient(16), 1.563) def test_juste_en_dessous_ancre_89(self): """89 m² est juste sous l'ancre neutre (90) → légèrement au-dessus de 1.0.""" # 1.0 + 0.75 * log10(90/89) = 1.00364 → 1.004 self.assertEqual(get_surface_coefficient(89), 1.004) def test_juste_au_dessus_ancre_91(self): """91 m² est juste au-dessus de l'ancre neutre (90) → légèrement sous 1.0.""" # 1.0 + 0.75 * log10(90/91) = 0.99642 → 0.996 self.assertEqual(get_surface_coefficient(91), 0.996) def test_premiere_surface_clampee_au_plancher(self): """À ~227 m², la formule passe sous 0.7 → clampée au plancher.""" # 1.0 + 0.75 * log10(90/227) = 0.6987 → max(0.7, ...) = 0.7 self.assertEqual(get_surface_coefficient(227), 0.7) def test_surface_float_non_entiere_ancre(self): """Surface flottante exactement à l'ancre → coefficient neutre 1.0.""" self.assertEqual(get_surface_coefficient(90.0), 1.0) def test_monotone_decroissant_apres_15(self): """Le coefficient doit décroître strictement quand la surface augmente (au-delà de 15).""" surfaces = [16, 20, 30, 50, 65, 90, 120, 175, 220] coefs = [get_surface_coefficient(s) for s in surfaces] assert coefs == sorted(coefs, reverse=True) class TestNormaliserCommuneLimites: def test_chaine_vide(self): assert _normaliser_commune("") == "" def test_chiffres_dans_le_nom_conserves(self): """Les chiffres ne sont pas filtrés par la normalisation.""" assert _normaliser_commune("Bois-73") == "BOIS 73" def test_espaces_initiaux_et_finaux_supprimes(self): assert _normaliser_commune(" Lille ") == "LILLE" def test_uniquement_tirets_devient_vide(self): """Une chaîne de tirets seulement doit être réduite à vide après strip.""" assert _normaliser_commune("---") == "" def test_triple_tiret_laisse_double_espace_interne(self): """Trois tirets consécutifs → trois espaces ; le remplacement simple en laisse deux.""" result = _normaliser_commune("A---B") assert " " in result # limitation du remplacement à passage unique class TestHaversineLimites: def test_points_quasi_antipodiaux(self): """Deux points opposés sur l'équateur → environ 20 015 km.""" dist = _haversine(0.0, 0.0, 0.0, 180.0) assert 19_000 < dist < 21_000 def test_distance_tres_courte(self): """Points séparés de ~1 m → distance positive et inférieure à 0.01 km.""" # 0.000009° de latitude ≈ 1 m dist = _haversine(50.0, 3.0, 50.000009, 3.0) assert 0.0 < dist < 0.01 def test_deplacement_purement_longitudinal(self): """Déplacement E-O à latitude fixe → distance positive.""" dist = _haversine(48.0, 2.0, 48.0, 3.0) assert dist > 0.0 def test_deplacement_purement_latitudinal(self): """Déplacement N-S à longitude fixe → distance positive.""" dist = _haversine(48.0, 2.0, 49.0, 2.0) assert dist > 0.0 def test_un_degre_latitude_environ_111km(self): """1° de latitude ≈ 111 km (rayon terrestre moyen).""" dist = _haversine(48.0, 2.0, 49.0, 2.0) assert 110 < dist < 112 class TestEvolutionCompoundingLimites: def _compound(self, annee_vente: int, evol_index: dict) -> float: compound = 1.0 for y in range(annee_vente + 1, _ANNEE_CIBLE + 1): evol = evol_index.get(("59290", y)) if evol is not None: compound *= 1.0 + evol / CENT return compound def test_evolution_zero_pct_est_neutre(self): """Une évolution de 0 % n'affecte pas le compound.""" evol_index = {("59290", 2025): 0.0} assert self._compound(2024, evol_index) == pytest.approx(1.0) def test_evolution_cent_pct_positif_double(self): """+100 % d'évolution → facteur ×2.""" evol_index = {("59290", 2025): 100.0} assert self._compound(2024, evol_index) == pytest.approx(2.0) def test_evolution_cent_pct_negatif_annule(self): """-100 % annule complètement le prix (compound = 0).""" evol_index = {("59290", 2025): -100.0} assert self._compound(2024, evol_index) == pytest.approx(0.0) def test_code_postal_different_ignore(self): """L'évolution d'un CP tiers ne doit pas affecter le compound du CP cible.""" evol_index = {("59290", 2025): 5.0, ("59000", 2025): 99.0} assert self._compound(2024, evol_index) == pytest.approx(1.05) def test_vente_annee_cible_aucun_compound(self): """Une vente de l'année cible elle-même → range vide → compound = 1.""" evol_index = {("59290", _ANNEE_CIBLE): 50.0} assert self._compound(_ANNEE_CIBLE, evol_index) == pytest.approx(1.0) class TestOutlierFilterLimites: def test_exactement_seuil_bas_inclus(self): """PRIX_M2_MIN lui-même est inclus dans la plage valide.""" assert PRIX_M2_MIN <= PRIX_M2_MIN <= PRIX_M2_MAX def test_exactement_seuil_haut_inclus(self): """PRIX_M2_MAX lui-même est inclus dans la plage valide.""" assert PRIX_M2_MIN <= PRIX_M2_MAX <= PRIX_M2_MAX def test_un_centime_sous_le_seuil_bas_exclu(self): assert not (PRIX_M2_MIN <= PRIX_M2_MIN - 0.01 <= PRIX_M2_MAX) def test_un_centime_au_dessus_du_seuil_haut_exclu(self): assert not (PRIX_M2_MIN <= PRIX_M2_MAX + 0.01 <= PRIX_M2_MAX) class TestRecencyWeightsLimites: def test_toutes_les_annees_dvf_couvertes(self): from domain.tools.analysis import _RECENCY_WEIGHTS, _ANNEES_DVF assert set(_RECENCY_WEIGHTS.keys()) == set(_ANNEES_DVF) def test_tous_les_poids_entiers_strictement_positifs(self): from domain.tools.analysis import _RECENCY_WEIGHTS assert all(isinstance(p, int) and p > 0 for p in _RECENCY_WEIGHTS.values()) def test_annee_la_plus_recente_a_le_poids_maximal(self): from domain.tools.analysis import _RECENCY_WEIGHTS assert _RECENCY_WEIGHTS[max(_RECENCY_WEIGHTS)] == max(_RECENCY_WEIGHTS.values()) class TestAnalyserBienLimites: async def test_sans_evolution_connue_compound_neutre(self, patched_analysis, monkeypatch): """Sans données prix_evolution, compound=1 pour toutes les ventes → evol_marche=0%.""" monkeypatch.setattr(patched_analysis, "_prix_evolution", []) monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) for ref in result["transactions_reference"]: assert ref["evol_marche_pct"] == pytest.approx(0.0, abs=0.01) async def test_adresse_geocodee_au_meme_endroit_que_transaction(self, patched_analysis, monkeypatch): """Adresse coïncidant exactement avec une transaction → DIST_EPSILON_KM évite la division par zéro.""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.671, 3.131)), # coord identique à la 1ère transaction sample ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) assert result["estimation"]["prix_m2_marche"] > 0 async def test_statut_terrible_prix_positif(self, patched_analysis, monkeypatch): """Même avec le pire statut (coef=0.50), le prix estimé reste positif.""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "terrible", "appartement", "est", ) assert result["estimation"]["prix_total_estime"] > 0 assert result["estimation"]["coef_renovation"] == pytest.approx(0.50) async def test_coef_orientation_nord_dans_resultat(self, patched_analysis, monkeypatch): """L'orientation nord (coef=0.90) est bien retournée dans le dict estimation.""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "nord", ) assert result["estimation"]["coef_orientation"] == pytest.approx(0.90) async def test_prix_total_triple_quand_surface_triple(self, patched_analysis, monkeypatch): """Le prix total est linéaire en surface (ratio 3× pour 30→90 m²).""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) r30 = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 30.0, "renovated", "appartement", "est", ) r90 = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 90.0, "renovated", "appartement", "est", ) ratio = r90["estimation"]["prix_total_estime"] / r30["estimation"]["prix_total_estime"] assert ratio == pytest.approx(3.0, rel=0.01) async def test_google_maps_url_contient_adresse(self, patched_analysis, monkeypatch): """L'URL Google Maps racine contient l'adresse encodée.""" monkeypatch.setattr( patched_analysis, "_geocode_adresse", AsyncMock(return_value=(50.672, 3.131)), ) result = await patched_analysis.analyser_bien( "12 rue de la paix", 59290, "wasquehal", 65.0, "renovated", "appartement", "est", ) assert "google.com/maps" in result["google_maps_url"]