spb/api-ka

1# ============================================2# Projet   : API-KA3# Fichier  : tests/test_connector_health.py4# Node     : m3u96b5# Author   : Simon-Pierre Boucher6# Contact  : contact@spboucher.ai7# Date     : 2026-08-188# ============================================9"""Tests de la supervision des connecteurs : classement d'état (fonctions pures),10parsing des /api/stats, upsert + anti-spam des alertes, endpoints publics."""1112from __future__ import annotations1314import datetime1516import pytest17from fastapi.testclient import TestClient18from sqlalchemy import delete, select1920from src.api.main import app21from src.database.db import session_scope22from src.database.models import ConnectorHealth23from src.monitoring import connector_health as ch24from src.monitoring.connector_health import (25    APP_SOURCE,26    SourceState,27    SyncEntry,28    assess_source,29    assess_unreachable,30    build_app_entry,31    parse_sync_entries,32    rotation_factor,33    run_connector_health_check,34)3536NOW = 1_787_040_000.037HOUR = 3600.0383940def _entry(hours_ago: float, found: int = 100, ok: bool = True, message: str = "") -> SyncEntry:41    return SyncEntry(ts=NOW - hours_ago * HOUR, found=found, ok=ok, message=message)424344# ---------------------------------------------------------------- classement454647def test_assess_ok_nominal():48    result = assess_source(49        [_entry(0.5, found=100), _entry(1.5, found=110)],50        SourceState(),51        NOW,52        stale_after_seconds=4 * HOUR,53    )54    assert result.status == "ok"55    assert result.found_last == 10056    assert result.consecutive_failures == 057    assert result.last_success is not None585960def test_assess_broken_apres_trois_echecs():61    entries = [62        _entry(0.5, found=0, ok=False, message="timeout"),63        _entry(1.5, found=0, ok=False),64        _entry(2.5, found=0, ok=False),65    ]66    result = assess_source(entries, SourceState(), NOW, stale_after_seconds=100 * HOUR)67    assert result.status == "broken"68    assert result.consecutive_failures == 3697071def test_assess_broken_apres_trois_zero_resultats():72    # Syncs "réussis" mais vides ×3 ALORS QUE la source a normalement des73    # résultats (médiane historique ≥ 1) → broken.74    entries = [_entry(h, found=0, ok=True) for h in (0.5, 1.5, 2.5)]75    prev = SourceState(median_found=40.0)76    result = assess_source(entries, prev, NOW, stale_after_seconds=100 * HOUR)77    assert result.status == "broken"78    assert result.consecutive_failures == 3798081def test_assess_zero_legitime_pas_broken():82    # Source vide légitimement (médiane historique < 1) : des syncs ok à83    # 0 résultat ne sont PAS des pannes (revue 2026-08-23 : niddamour,84    # courtemanche, place_florimay, atlas_immo — vérifiés à la main).85    entries = [_entry(h, found=0, ok=True) for h in (0.5, 1.5, 2.5)]86    result = assess_source(87        entries, SourceState(median_found=0.0), NOW, stale_after_seconds=100 * HOUR88    )89    assert result.status != "broken"90    assert result.consecutive_failures == 0919293def test_assess_zero_legitime_pas_stale():94    # Une source vide légitime qui synchronise ok n'est pas « stale » même95    # si son dernier succès AVEC résultats est ancien.96    prev = SourceState(97        median_found=0.0,98        last_success=datetime.datetime.fromtimestamp(99            NOW - 500 * HOUR, tz=datetime.UTC100        ),101    )102    entries = [_entry(0.5, found=0, ok=True)]103    result = assess_source(entries, prev, NOW, stale_after_seconds=100 * HOUR)104    assert result.status not in ("stale", "broken")105106107def test_assess_streak_cumule_avec_etat_precedent():108    prev = SourceState(109        status="degraded",110        consecutive_failures=2,111        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),112    )113    # Une seule nouvelle entrée en échec suffit alors pour atteindre 3.114    result = assess_source(115        [_entry(0.5, found=0, ok=False)], prev, NOW, stale_after_seconds=100 * HOUR116    )117    assert result.status == "broken"118    assert result.consecutive_failures == 3119120121def test_assess_succes_remet_le_streak_a_zero():122    prev = SourceState(123        consecutive_failures=2,124        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),125    )126    result = assess_source(127        [_entry(0.5, found=80)], prev, NOW, stale_after_seconds=100 * HOUR128    )129    assert result.status == "ok"130    assert result.consecutive_failures == 0131132133def test_assess_stale_sans_sync_recent():134    prev = SourceState(135        status="ok",136        last_success=datetime.datetime.fromtimestamp(NOW - 10 * HOUR, tz=datetime.UTC),137        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),138        median_found=100.0,139        found_last=100,140    )141    # Cadence 1 h → stale au-delà de 2 h sans succès.142    result = assess_source([], prev, NOW, stale_after_seconds=2 * HOUR)143    assert result.status == "stale"144145146def test_assess_degraded_si_volume_sous_la_moitie_de_la_mediane():147    prev = SourceState(148        median_found=200.0,149        last_success=datetime.datetime.fromtimestamp(NOW - 1 * HOUR, tz=datetime.UTC),150    )151    result = assess_source(152        [_entry(0.5, found=40)], prev, NOW, stale_after_seconds=100 * HOUR153    )154    assert result.status == "degraded"155    assert "50 %" in result.message156157158def test_assess_priorite_broken_avant_stale():159    prev = SourceState(160        last_success=datetime.datetime.fromtimestamp(NOW - 50 * HOUR, tz=datetime.UTC),161        consecutive_failures=5,162        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),163    )164    result = assess_source([], prev, NOW, stale_after_seconds=2 * HOUR)165    assert result.status == "broken"166167168def test_assess_unreachable_degrade_puis_broken():169    first = assess_unreachable(SourceState(), "connexion refusée")170    assert first.status == "degraded"171    assert first.consecutive_failures == 1172    third = assess_unreachable(173        SourceState(consecutive_failures=2), "connexion refusée"174    )175    assert third.status == "broken"176177178# ------------------------------------------------------------------- parsing179180181def test_parse_recent_syncs_et_sync_log():182    payload = {183        "recent_syncs": [184            {"source": "kangalou", "ts": NOW - 100, "found": 7916, "ok": 1, "message": "ok"},185            {"source": "kangalou", "ts": NOW - 4000, "found": 0, "ok": 0, "message": "err"},186        ]187    }188    by_source = parse_sync_entries(payload)189    assert list(by_source) == ["kangalou"]190    assert len(by_source["kangalou"]) == 2191    assert by_source["kangalou"][1].ok is False192193    fabrika = {194        "sync_log": [195            {"ts": NOW - 50, "store_id": "daigneau.ca", "found": 25, "status": "ok"},196            {"ts": NOW - 60, "store_id": "x.com", "found": 0, "status": "error"},197        ]198    }199    by_store = parse_sync_entries(fabrika)200    assert by_store["daigneau.ca"][0].ok is True201    assert by_store["x.com"][0].ok is False202203204def test_build_app_entry_last_sync_et_fallback():205    # crea-ka : last_sync global ISO.206    creaka = {"creators": 4882, "last_sync": "2026-08-18T06:27:51Z"}207    entry = build_app_entry("creaka", creaka, {}, NOW)208    assert entry.ok is True209    assert entry.found == 4882210    assert entry.ts != NOW211212    # sorti-ka : aucun horodatage → joignabilité + volume total.213    sortika = {"total_active": 15820, "sources": 10}214    entry = build_app_entry("sortika", sortika, {}, NOW)215    assert entry.ts == NOW216    assert entry.found == 15820217218219def test_rotation_factor():220    by_source = {f"s{i}": [_entry(1)] for i in range(20)}221    assert rotation_factor({"sources": 206}, by_source) == 11222    assert rotation_factor({"sources": 3}, by_source) == 1223    assert rotation_factor({}, {}) == 1224225226# ------------------------------------------- job complet : upsert + anti-spam227228229@pytest.fixture()230def clean_connector_table():231    with session_scope() as session:232        session.execute(delete(ConnectorHealth))233    yield234    with session_scope() as session:235        session.execute(delete(ConnectorHealth))236237238def _fake_louka_payload(ok: bool) -> dict:239    return {240        "total": 44343,241        "sources": 2,242        "recent_syncs": [243            {244                "source": "kangalou",245                "ts": NOW - 600,246                "found": 0 if not ok else 7916,247                "ok": 0 if not ok else 1,248                "message": "err" if not ok else "ok",249            }250        ],251    }252253254def test_job_upsert_et_alerte_sans_repetition(255    monkeypatch: pytest.MonkeyPatch, clean_connector_table, isolated_logs_dir256):257    """3 runs en échec → broken + UNE alerte ; run suivant identique → pas de258    nouvelle alerte (anti-spam) ; retour au succès → ok."""259    alerts_file = isolated_logs_dir / "alerts.log"260    baseline = alerts_file.read_text() if alerts_file.exists() else ""261262    monkeypatch.setattr(ch, "_stats_url", lambda service: "http://test/api/stats")263    payload = {"value": _fake_louka_payload(ok=False)}264    monkeypatch.setattr(ch, "_fetch_stats", lambda url: payload["value"])265266    # Trois runs espacés de 2 h, chacun voyant UNE nouvelle synchro en échec267    # (une entrée déjà comptée — ts <= checked_at précédent — ne recompte pas).268    for i in range(3):269        run_at = NOW + i * 2 * HOUR270        p = _fake_louka_payload(ok=False)271        p["recent_syncs"][0]["ts"] = run_at - 600272        payload["value"] = p273        ch._check_service("louka", run_at)274275    with session_scope() as session:276        row = session.execute(277            select(ConnectorHealth).where(278                ConnectorHealth.service == "louka",279                ConnectorHealth.source == "kangalou",280            )281        ).scalar_one()282        assert row.status == "broken"283        assert row.consecutive_failures == 3284285    content = alerts_file.read_text()[len(baseline):]286    assert content.count("[connecteurs] louka/kangalou : broken") == 1287288    # Même état au run suivant → aucune nouvelle alerte.289    run_at = NOW + 3 * 2 * HOUR290    p = _fake_louka_payload(ok=False)291    p["recent_syncs"][0]["ts"] = run_at - 600292    payload["value"] = p293    ch._check_service("louka", run_at)294    content = alerts_file.read_text()[len(baseline):]295    assert content.count("[connecteurs] louka/kangalou : broken") == 1296297    # Retour au succès → ok, streak remis à zéro.298    run_at = NOW + 4 * 2 * HOUR299    p = _fake_louka_payload(ok=True)300    p["recent_syncs"][0]["ts"] = run_at - 600301    payload["value"] = p302    ch._check_service("louka", run_at)303    with session_scope() as session:304        row = session.execute(305            select(ConnectorHealth).where(306                ConnectorHealth.service == "louka",307                ConnectorHealth.source == "kangalou",308            )309        ).scalar_one()310        assert row.status == "ok"311        assert row.consecutive_failures == 0312313314def test_job_app_injoignable_ne_crashe_pas(315    monkeypatch: pytest.MonkeyPatch, clean_connector_table316):317    monkeypatch.setattr(ch, "_stats_url", lambda service: "http://test/api/stats")318319    def _boom(url):320        raise ConnectionError("refusée")321322    monkeypatch.setattr(ch, "_fetch_stats", _boom)323    summary = run_connector_health_check(NOW)324    assert len(summary["services"]) == 9325    with session_scope() as session:326        row = session.execute(327            select(ConnectorHealth).where(328                ConnectorHealth.service == "louka",329                ConnectorHealth.source == APP_SOURCE,330            )331        ).scalar_one()332        assert row.status == "degraded"333        assert row.consecutive_failures == 1334335336# ----------------------------------------------------------------- endpoints337338339def test_endpoints_monitoring(clean_connector_table):340    now = datetime.datetime.now(tz=datetime.UTC)341    with session_scope() as session:342        session.add(343            ConnectorHealth(344                service="louka",345                source=APP_SOURCE,346                checked_at=now,347                status="ok",348                last_success=now,349                found_last=44343,350                median_found=44000.0,351                consecutive_failures=0,352                message="ok",353            )354        )355        session.add(356            ConnectorHealth(357                service="louka",358                source="kangalou",359                checked_at=now,360                status="broken",361                last_success=None,362                found_last=0,363                median_found=7900.0,364                consecutive_failures=3,365                message="3 synchros consécutives en échec ou à 0 résultat",366            )367        )368369    client = TestClient(app)370371    body = client.get("/api/v1/monitoring/connectors").json()372    assert body["success"] is True373    assert body["data"]["summary"]["ok"] == 1374    assert body["data"]["summary"]["broken"] == 1375    louka = body["data"]["services"]["louka"]376    assert louka["app"]["status"] == "ok"377    assert louka["connectors"][0]["source"] == "kangalou"378379    body = client.get("/api/v1/monitoring/connectors/louka").json()380    assert body["data"]["service"] == "louka"381    assert body["data"]["summary"]["broken"] == 1382383    assert client.get("/api/v1/monitoring/connectors/nimporte").status_code == 404384385    # Résumé intégré au /health.386    body = client.get("/health").json()387    assert body["data"]["connectors"] == {388        "ok": 1,389        "degraded": 0,390        "broken": 1,391        "stale": 0,392        "retired": 0,393    }394