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 → broken.73    entries = [_entry(h, found=0, ok=True) for h in (0.5, 1.5, 2.5)]74    result = assess_source(entries, SourceState(), NOW, stale_after_seconds=100 * HOUR)75    assert result.status == "broken"76    assert result.consecutive_failures == 3777879def test_assess_streak_cumule_avec_etat_precedent():80    prev = SourceState(81        status="degraded",82        consecutive_failures=2,83        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),84    )85    # Une seule nouvelle entrée en échec suffit alors pour atteindre 3.86    result = assess_source(87        [_entry(0.5, found=0, ok=False)], prev, NOW, stale_after_seconds=100 * HOUR88    )89    assert result.status == "broken"90    assert result.consecutive_failures == 3919293def test_assess_succes_remet_le_streak_a_zero():94    prev = SourceState(95        consecutive_failures=2,96        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),97    )98    result = assess_source(99        [_entry(0.5, found=80)], prev, NOW, stale_after_seconds=100 * HOUR100    )101    assert result.status == "ok"102    assert result.consecutive_failures == 0103104105def test_assess_stale_sans_sync_recent():106    prev = SourceState(107        status="ok",108        last_success=datetime.datetime.fromtimestamp(NOW - 10 * HOUR, tz=datetime.UTC),109        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),110        median_found=100.0,111        found_last=100,112    )113    # Cadence 1 h → stale au-delà de 2 h sans succès.114    result = assess_source([], prev, NOW, stale_after_seconds=2 * HOUR)115    assert result.status == "stale"116117118def test_assess_degraded_si_volume_sous_la_moitie_de_la_mediane():119    prev = SourceState(120        median_found=200.0,121        last_success=datetime.datetime.fromtimestamp(NOW - 1 * HOUR, tz=datetime.UTC),122    )123    result = assess_source(124        [_entry(0.5, found=40)], prev, NOW, stale_after_seconds=100 * HOUR125    )126    assert result.status == "degraded"127    assert "50 %" in result.message128129130def test_assess_priorite_broken_avant_stale():131    prev = SourceState(132        last_success=datetime.datetime.fromtimestamp(NOW - 50 * HOUR, tz=datetime.UTC),133        consecutive_failures=5,134        checked_at=datetime.datetime.fromtimestamp(NOW - 2 * HOUR, tz=datetime.UTC),135    )136    result = assess_source([], prev, NOW, stale_after_seconds=2 * HOUR)137    assert result.status == "broken"138139140def test_assess_unreachable_degrade_puis_broken():141    first = assess_unreachable(SourceState(), "connexion refusée")142    assert first.status == "degraded"143    assert first.consecutive_failures == 1144    third = assess_unreachable(145        SourceState(consecutive_failures=2), "connexion refusée"146    )147    assert third.status == "broken"148149150# ------------------------------------------------------------------- parsing151152153def test_parse_recent_syncs_et_sync_log():154    payload = {155        "recent_syncs": [156            {"source": "kangalou", "ts": NOW - 100, "found": 7916, "ok": 1, "message": "ok"},157            {"source": "kangalou", "ts": NOW - 4000, "found": 0, "ok": 0, "message": "err"},158        ]159    }160    by_source = parse_sync_entries(payload)161    assert list(by_source) == ["kangalou"]162    assert len(by_source["kangalou"]) == 2163    assert by_source["kangalou"][1].ok is False164165    fabrika = {166        "sync_log": [167            {"ts": NOW - 50, "store_id": "daigneau.ca", "found": 25, "status": "ok"},168            {"ts": NOW - 60, "store_id": "x.com", "found": 0, "status": "error"},169        ]170    }171    by_store = parse_sync_entries(fabrika)172    assert by_store["daigneau.ca"][0].ok is True173    assert by_store["x.com"][0].ok is False174175176def test_build_app_entry_last_sync_et_fallback():177    # crea-ka : last_sync global ISO.178    creaka = {"creators": 4882, "last_sync": "2026-08-18T06:27:51Z"}179    entry = build_app_entry("creaka", creaka, {}, NOW)180    assert entry.ok is True181    assert entry.found == 4882182    assert entry.ts != NOW183184    # sorti-ka : aucun horodatage → joignabilité + volume total.185    sortika = {"total_active": 15820, "sources": 10}186    entry = build_app_entry("sortika", sortika, {}, NOW)187    assert entry.ts == NOW188    assert entry.found == 15820189190191def test_rotation_factor():192    by_source = {f"s{i}": [_entry(1)] for i in range(20)}193    assert rotation_factor({"sources": 206}, by_source) == 11194    assert rotation_factor({"sources": 3}, by_source) == 1195    assert rotation_factor({}, {}) == 1196197198# ------------------------------------------- job complet : upsert + anti-spam199200201@pytest.fixture()202def clean_connector_table():203    with session_scope() as session:204        session.execute(delete(ConnectorHealth))205    yield206    with session_scope() as session:207        session.execute(delete(ConnectorHealth))208209210def _fake_louka_payload(ok: bool) -> dict:211    return {212        "total": 44343,213        "sources": 2,214        "recent_syncs": [215            {216                "source": "kangalou",217                "ts": NOW - 600,218                "found": 0 if not ok else 7916,219                "ok": 0 if not ok else 1,220                "message": "err" if not ok else "ok",221            }222        ],223    }224225226def test_job_upsert_et_alerte_sans_repetition(227    monkeypatch: pytest.MonkeyPatch, clean_connector_table, isolated_logs_dir228):229    """3 runs en échec → broken + UNE alerte ; run suivant identique → pas de230    nouvelle alerte (anti-spam) ; retour au succès → ok."""231    alerts_file = isolated_logs_dir / "alerts.log"232    baseline = alerts_file.read_text() if alerts_file.exists() else ""233234    monkeypatch.setattr(ch, "_stats_url", lambda service: "http://test/api/stats")235    payload = {"value": _fake_louka_payload(ok=False)}236    monkeypatch.setattr(ch, "_fetch_stats", lambda url: payload["value"])237238    # Trois runs espacés de 2 h, chacun voyant UNE nouvelle synchro en échec239    # (une entrée déjà comptée — ts <= checked_at précédent — ne recompte pas).240    for i in range(3):241        run_at = NOW + i * 2 * HOUR242        p = _fake_louka_payload(ok=False)243        p["recent_syncs"][0]["ts"] = run_at - 600244        payload["value"] = p245        ch._check_service("louka", run_at)246247    with session_scope() as session:248        row = session.execute(249            select(ConnectorHealth).where(250                ConnectorHealth.service == "louka",251                ConnectorHealth.source == "kangalou",252            )253        ).scalar_one()254        assert row.status == "broken"255        assert row.consecutive_failures == 3256257    content = alerts_file.read_text()[len(baseline):]258    assert content.count("[connecteurs] louka/kangalou : broken") == 1259260    # Même état au run suivant → aucune nouvelle alerte.261    run_at = NOW + 3 * 2 * HOUR262    p = _fake_louka_payload(ok=False)263    p["recent_syncs"][0]["ts"] = run_at - 600264    payload["value"] = p265    ch._check_service("louka", run_at)266    content = alerts_file.read_text()[len(baseline):]267    assert content.count("[connecteurs] louka/kangalou : broken") == 1268269    # Retour au succès → ok, streak remis à zéro.270    run_at = NOW + 4 * 2 * HOUR271    p = _fake_louka_payload(ok=True)272    p["recent_syncs"][0]["ts"] = run_at - 600273    payload["value"] = p274    ch._check_service("louka", run_at)275    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 == "ok"283        assert row.consecutive_failures == 0284285286def test_job_app_injoignable_ne_crashe_pas(287    monkeypatch: pytest.MonkeyPatch, clean_connector_table288):289    monkeypatch.setattr(ch, "_stats_url", lambda service: "http://test/api/stats")290291    def _boom(url):292        raise ConnectionError("refusée")293294    monkeypatch.setattr(ch, "_fetch_stats", _boom)295    summary = run_connector_health_check(NOW)296    assert len(summary["services"]) == 9297    with session_scope() as session:298        row = session.execute(299            select(ConnectorHealth).where(300                ConnectorHealth.service == "louka",301                ConnectorHealth.source == APP_SOURCE,302            )303        ).scalar_one()304        assert row.status == "degraded"305        assert row.consecutive_failures == 1306307308# ----------------------------------------------------------------- endpoints309310311def test_endpoints_monitoring(clean_connector_table):312    now = datetime.datetime.now(tz=datetime.UTC)313    with session_scope() as session:314        session.add(315            ConnectorHealth(316                service="louka",317                source=APP_SOURCE,318                checked_at=now,319                status="ok",320                last_success=now,321                found_last=44343,322                median_found=44000.0,323                consecutive_failures=0,324                message="ok",325            )326        )327        session.add(328            ConnectorHealth(329                service="louka",330                source="kangalou",331                checked_at=now,332                status="broken",333                last_success=None,334                found_last=0,335                median_found=7900.0,336                consecutive_failures=3,337                message="3 synchros consécutives en échec ou à 0 résultat",338            )339        )340341    client = TestClient(app)342343    body = client.get("/api/v1/monitoring/connectors").json()344    assert body["success"] is True345    assert body["data"]["summary"]["ok"] == 1346    assert body["data"]["summary"]["broken"] == 1347    louka = body["data"]["services"]["louka"]348    assert louka["app"]["status"] == "ok"349    assert louka["connectors"][0]["source"] == "kangalou"350351    body = client.get("/api/v1/monitoring/connectors/louka").json()352    assert body["data"]["service"] == "louka"353    assert body["data"]["summary"]["broken"] == 1354355    assert client.get("/api/v1/monitoring/connectors/nimporte").status_code == 404356357    # Résumé intégré au /health.358    body = client.get("/health").json()359    assert body["data"]["connectors"] == {360        "ok": 1,361        "degraded": 0,362        "broken": 1,363        "stale": 0,364    }365