Steph680/ormuz

1# -*- coding: utf-8 -*-2"""Tests de la fenêtre IA (récupération + prompt, sans réseau)."""3 4import asyncio5 6from app import ask, db7from app.state import utc_now8 9 10def test_extraction_mots_cles():11    kw = ask.keywords("Quels pétroliers ont été attaqués dans le détroit ?")12    assert "pétroliers" in kw and "attaqués" in kw13    # Mots vides et thème générique écartés.14    assert "dans" not in kw and "ormuz" not in kw and "hormuz" not in kw15 16 17def test_contexte_et_prompt(tmp_path, monkeypatch):18    monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))19 20    async def scenario():21        await db.init_db()22        async with db.connect() as conn:23            await conn.execute(24                "INSERT INTO news_events (url, published_at, title, title_fr,"25                " source, lang, category, relevance, classified_at)"26                " VALUES ('u1', '2026-09-05T10:00:00', 'US sinks Iranian tanker',"27                " 'Les USA coulent un pétrolier iranien', 'Reuters', 'en',"28                " 'incident_securite', 90, '2026-09-05')")29            await conn.execute(30                "INSERT INTO news_events (url, published_at, title, source, lang,"31                " category, relevance, classified_at)"32                " VALUES ('u2', '2026-09-04T10:00:00', 'Football result',"33                " 'ESPN', 'en', 'autre', 10, '2026-09-04')")34            await conn.execute(35                "INSERT INTO geo_incidents (url, lat, lon, source, type, place,"36                " published_at, title) VALUES ('g1', 26.4, 56.5, 'UKMTO',"37                " 'Attack', 'Strait of Hormuz', '2026-09-05T09:00:00', 'Tanker hit')")38            await conn.commit()39 40        ctx = await ask.gather_context("pétrolier coulé", "fr")41        titres = [e["t"] for e in ctx["events"]]42        # L'événement pertinent est retenu ; le bruit 'autre' est exclu.43        assert any("pétrolier" in x for x in titres)44        assert all("Football" not in x for x in titres)45        assert len(ctx["incidents"]) == 146 47        messages = ask.build_prompt("pétrolier coulé ?", ctx, "fr")48        assert messages[0]["role"] == "system"49        blob = messages[1]["content"]50        assert "pétrolier" in blob and "UKMTO" in blob51        # Consigne d'ancrage stricte présente.52        assert "EXCLUSIVEMENT" in messages[0]["content"]53 54    asyncio.run(scenario())55 56 57def test_question_trop_courte(tmp_path, monkeypatch):58    monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))59    monkeypatch.setenv("LLM_API_KEY", "x")60 61    async def scenario():62        await db.init_db()63        try:64            await ask.answer("?", "fr")65            assert False, "aurait dû lever"66        except ValueError:67            pass68 69    asyncio.run(scenario())70