Steph680/ormuz
0
1"""Tests du rapport d'incident (LLM simulé, sans réseau)."""2 3import asyncio4from datetime import date5 6from app import db, rapport7 8 9def test_format_degres_minutes_diziemes():10 assert rapport.format_dm(25.7, 56.6) == "25°42.0'N 056°36.0'E"11 assert rapport.format_dm(26.5083, 56.3467) == "26°30.5'N 056°20.8'E"12 assert rapport.format_dm(-15.0, -42.5) == "15°00.0'S 042°30.0'W"13 # Arrondi 59.96' -> degré suivant, jamais « 60.0' ».14 assert rapport.format_dm(25.9993, 56.0) == "26°00.0'N 056°00.0'E"15 16 17async def _seed(conn):18 rows = [19 # L'alerte officielle de départ.20 ("https://ukmto/88", "2026-07-17T06:00:00+00:00",21 "UKMTO Attack 88 — Strait of Hormuz : merchant vessel struck by projectile",22 "UKMTO", "incident_securite", 85),23 # Presse : même événement (mots partagés), dans la fenêtre.24 ("https://p/1", "2026-07-17T12:00:00+00:00",25 "Merchant vessel struck by projectile in Strait of Hormuz, crew safe",26 "Reuters", "incident_securite", 80),27 # Presse : autre sujet, même fenêtre -> écarté (pas assez de mots partagés).28 ("https://p/2", "2026-07-16T09:00:00+00:00",29 "Iran threatens to close the strait after sanctions",30 "AFP", "declaration_iran", 70),31 # Presse : même événement mais hors fenêtre de ±2 jours -> écarté.32 ("https://p/3", "2026-07-25T09:00:00+00:00",33 "Merchant vessel struck by projectile: investigation continues",34 "AP", "incident_securite", 75),35 ]36 for url, pub, title, src, cat, rel in rows:37 await conn.execute(38 "INSERT INTO news_events (url, published_at, title, source, category, relevance)"39 " VALUES (?, ?, ?, ?, ?, ?)", (url, pub, title, src, cat, rel))40 await conn.execute(41 "INSERT INTO geo_incidents (url, lat, lon, source, type, place, published_at, title)"42 " VALUES ('https://ukmto/88', 26.5, 56.4, 'UKMTO', 'Attack', 'Strait of Hormuz',"43 " '2026-07-17T06:00:00+00:00', 'UKMTO Attack 88')")44 await conn.commit()45 46 47def test_contexte_rapproche_la_bonne_presse(tmp_path, monkeypatch):48 monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))49 50 async def scenario():51 await db.init_db()52 async with db.connect() as conn:53 await _seed(conn)54 cur = await conn.execute("SELECT id, source FROM news_events ORDER BY id")55 ids = {src: i for i, src in await cur.fetchall()}56 57 ctx = await rapport.build_context(ids["UKMTO"])58 assert ctx["alerte"]["source"] == "UKMTO"59 assert ctx["position"]["lat"] == 26.560 sources_presse = [p["source"] for p in ctx["presse"]]61 assert sources_presse == ["Reuters"] # ni AFP (autre sujet) ni AP (hors fenêtre)62 63 # Un article de presse n'est pas un point de départ valide.64 assert await rapport.build_context(ids["Reuters"]) is None65 66 # Le prompt sépare bien alerte et presse.67 messages = rapport.build_prompt(ctx, "fr")68 corps = messages[1]["content"]69 assert "ALERTE OFFICIELLE" in corps and "Reuters" in corps70 assert "FAITS ÉTABLIS" in corps71 72 # Version anglaise : consignes et rubriques en anglais.73 messages_en = rapport.build_prompt(ctx, "en")74 assert "OFFICIAL ALERT" in messages_en[1]["content"]75 assert "ESTABLISHED FACTS" in messages_en[1]["content"]76 assert "must be in English" in messages_en[0]["content"]77 78 asyncio.run(scenario())79 80 81def test_generate_avec_cache(tmp_path, monkeypatch):82 monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))83 appels = []84 85 async def faux_llm(messages):86 appels.append(1)87 return "1. ÉVÉNEMENT\nAttaque confirmée.", "modele-test"88 89 monkeypatch.setattr(rapport, "_call_llm", faux_llm)90 91 async def scenario():92 await db.init_db()93 async with db.connect() as conn:94 await _seed(conn)95 cur = await conn.execute(96 "SELECT id FROM news_events WHERE source = 'UKMTO'")97 (uid,) = await cur.fetchone()98 99 r1 = await rapport.generate(uid, "fr")100 assert r1["cache"] is False and "Attaque confirmée" in r1["rapport"]101 r2 = await rapport.generate(uid, "fr")102 assert r2["cache"] is True and len(appels) == 1103 r3 = await rapport.generate(uid, "fr", force=True)104 assert r3["cache"] is False and len(appels) == 2105 assert await rapport.generate(99999, "fr") is None106 107 # Nouvelle presse rapprochée -> le cache est périmé, régénération auto.108 async with db.connect() as conn:109 await conn.execute(110 "INSERT INTO news_events (url, published_at, title, source,"111 " category, relevance) VALUES ('https://p/9',"112 " '2026-07-18T08:00:00+00:00', 'Merchant vessel struck by"113 " projectile: two injured reported', 'BBC', 'incident_securite', 80)")114 await conn.commit()115 avant = len(appels)116 r4 = await rapport.generate(uid, "fr")117 assert r4["cache"] is False and len(appels) == avant + 1118 119 asyncio.run(scenario())120 121 122def test_ancre_osint_acceptee(tmp_path, monkeypatch):123 monkeypatch.setenv("DB_PATH", str(tmp_path / "t.db"))124 125 async def scenario():126 await db.init_db()127 async with db.connect() as conn:128 await conn.execute(129 "INSERT INTO news_events (url, published_at, title, source,"130 " category, relevance) VALUES ('https://wiki#kylo',"131 " '2026-09-06T00:00:00+00:00', 'MT Kylo struck by US missile in"132 " Gulf of Oman', 'OSINT', 'incident_securite', 75)")133 await conn.execute(134 "INSERT INTO news_events (url, published_at, title, source,"135 " category, relevance) VALUES ('https://p/kylo',"136 " '2026-09-06T06:00:00+00:00', 'MT Kylo struck by missile in the"137 " Gulf of Oman, tanker ablaze', 'Reuters', 'incident_securite', 80)")138 await conn.commit()139 cur = await conn.execute("SELECT id FROM news_events WHERE source='OSINT'")140 (oid,) = await cur.fetchone()141 # Un incident OSINT est un point de départ valide, enrichi par la presse.142 ctx = await rapport.build_context(oid)143 assert ctx is not None and ctx["alerte"]["source"] == "OSINT"144 assert any(p["source"] == "Reuters" for p in ctx["presse"])145 146 asyncio.run(scenario())147