Steph680/ormuz

1"""Tests de l'auto-diagnostic (sondes simulées, sans réseau)."""2 3import asyncio4from datetime import timedelta5 6from app import db, selfcheck7from app.state import STATUS, utc_now8 9 10async def _jeton_ok():11    return {"nom": "Jeton Hugging Face", "niveau": "ok", "detail": "simulé"}12 13 14def test_verdict_global_est_le_pire(tmp_path, monkeypatch):15    monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))16    monkeypatch.setattr(selfcheck, "_check_jeton_hf", _jeton_ok)17    monkeypatch.setattr(STATUS, "gdelt_last_poll_at", utc_now())18    monkeypatch.setattr(STATUS, "gdelt_last_error", "")19    monkeypatch.setattr(STATUS, "backfill_state", "terminé")20    monkeypatch.setattr(STATUS, "translate_last_error", "")21    monkeypatch.setattr(STATUS, "translate_pending", 0)22 23    async def scenario():24        await db.init_db()25        # Aucune donnée PortWatch : la sonde trafic est rouge -> verdict global bad.26        result = await selfcheck.run_checks()27        assert result["verdict"] == "bad"28        par_nom = {s["nom"]: s["niveau"] for s in result["sondes"]}29        assert par_nom["Trafic PortWatch"] == "bad"30        assert par_nom["Base SQLite"] == "ok"31 32        # Avec du trafic frais, tout passe au vert (sauvegarde en mode dev = warn).33        async with db.connect() as conn:34            await conn.execute(35                "INSERT INTO portwatch_daily (date_utc, n_cargo, n_tanker, n_total)"36                " VALUES (?, 40, 30, 70)",37                ((utc_now().date() - timedelta(days=6)).isoformat(),))38            await conn.commit()39        result = await selfcheck.run_checks()40        par_nom = {s["nom"]: s["niveau"] for s in result["sondes"]}41        assert par_nom["Trafic PortWatch"] == "ok"42        assert result["verdict"] == "warn"  # sauvegarde non configurée en local43 44    asyncio.run(scenario())45 46 47def test_historique_stocke_et_purge(tmp_path, monkeypatch):48    monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))49 50    async def scenario():51        await db.init_db()52        # Un vieux verdict au-delà de 7 jours doit être purgé au prochain stockage.53        async with db.connect() as conn:54            await conn.execute(55                "INSERT INTO health_checks (checked_at, verdict, details) VALUES (?, 'ok', '[]')",56                ((utc_now() - timedelta(days=9)).isoformat(),))57            await conn.commit()58        await selfcheck.store_result(59            {"verifie_le": utc_now().isoformat(), "verdict": "warn", "sondes": []})60        hist = await selfcheck.history()61        assert len(hist) == 162        assert hist[0]["verdict"] == "warn"63 64    asyncio.run(scenario())65