"""Historique et courbes d'évolution (curseur SQL simulé via la fixture `cur`).""" def _requetes(cur) -> str: """Concatène les requêtes exécutées, pour vérifier le SQL construit.""" return " ".join(q for q, _ in cur.executed) def _parametres(cur) -> list: return [p for _, args in cur.executed for p in args] def test_historique_exige_authentification(client): assert client.get("/historique").status_code == 401 def test_historique_sans_filtre_ne_restreint_rien(client, cur, auth_headers): cur.description = [("id_monito",), ("nb_erreurs",)] cur._rows = [(1, 5)] r = client.get("/historique", headers=auth_headers) assert r.status_code == 200 requete = _requetes(cur) assert "date_sauvegarde >=" not in requete assert "date_sauvegarde <=" not in requete def test_historique_applique_les_bornes_de_dates(client, cur, auth_headers): cur.description = [("id_monito",)] cur._rows = [] r = client.get( "/historique", params={"date_debut": "2026-01-01", "date_fin": "2026-08-15"}, headers=auth_headers, ) assert r.status_code == 200 requete = _requetes(cur) assert "date_sauvegarde >= ?" in requete assert "date_sauvegarde <= ?" in requete # Les dates sont passées en paramètres, jamais concaténées dans le SQL. assert "2026-01-01" in _parametres(cur) assert "2026-08-15" in _parametres(cur) def test_historique_refuse_une_date_mal_formee(client, cur, auth_headers): r = client.get("/historique", params={"date_debut": "15/01/2026"}, headers=auth_headers) assert r.status_code == 422 def test_historique_filtre_par_monitoring(client, cur, auth_headers): cur.description = [("id_monito",)] cur._rows = [] client.get("/historique", params={"id_monito": 3}, headers=auth_headers) assert "id_monito = ?" in _requetes(cur) assert 3 in _parametres(cur) def test_evolution_monitoring_inconnu_renvoie_404(client, cur, auth_headers): # 999 n'est pas dans MONITO_TABLES : on refuse avant d'interroger la base. r = client.get("/historique/999/evolution", headers=auth_headers) assert r.status_code == 404 def test_evolution_formate_les_points_pour_le_graphique(client, cur, auth_headers): cur._rows = [("2026-01-01", 12), ("2026-01-02", 9)] r = client.get("/historique/1/evolution", headers=auth_headers) assert r.status_code == 200 corps = r.json() assert corps["points"] == 2 assert corps["evolution"][0] == {"date": "2026-01-01", "nb_erreurs": 12} def test_evolution_globale_renvoie_une_serie(client, cur, auth_headers): cur._rows = [("2026-01-01", 56), ("2026-01-02", 54)] r = client.get("/evolution/global", headers=auth_headers) assert r.status_code == 200 corps = r.json() assert corps["type"] == "GLOBAL" assert corps["points"] == 2 assert corps["series"][0]["nb_erreurs"] == 56 def test_evolution_par_monitoring_regroupe_par_courbe(client, cur, auth_headers): # Lignes à plat en base -> une série par monitoring côté API. cur._rows = [ ("2026-01-01", 1, "Tiers payeurs", 5), ("2026-01-02", 1, "Tiers payeurs", 4), ("2026-01-01", 2, "Sans RIB", 6), ] r = client.get("/evolution/par-monitoring", headers=auth_headers) assert r.status_code == 200 corps = r.json() assert corps["nb_series"] == 2 series = {m["id_monito"]: m for m in corps["monitorings"]} assert len(series[1]["serie"]) == 2 assert len(series[2]["serie"]) == 1 assert series[1]["nom_monito"] == "Tiers payeurs" def test_evolution_par_service_regroupe_par_service(client, cur, auth_headers): cur._rows = [ ("2026-01-01", "Contrat", 20), ("2026-01-02", "Contrat", 18), ("2026-01-01", "Fournisseur", 5), ] r = client.get("/evolution/par-service", headers=auth_headers) assert r.status_code == 200 corps = r.json() assert corps["nb_series"] == 2 par_service = {s["service"]: s["serie"] for s in corps["services"]} assert len(par_service["Contrat"]) == 2 assert len(par_service["Fournisseur"]) == 1