d81ea68dec
Stop the live HUD reconnect storm (nginx WS snippet + backoff), copy intel/nous_client into the summarizer image, and make event ingest idempotent on URL. GDELT uses the DOC API; NWS no longer sends bbox; FIRMS is one ON CONFLICT batch; GET /api/aircraft serves last-known. Health reports freshness without 503ing docker. EONET + CISA KEV added.
133 lines
4 KiB
Python
133 lines
4 KiB
Python
"""Generic event ingest: idempotency, USGS ids, GDELT DOC URL."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
def test_event_dedup_key_prefers_url():
|
|
from sources import event_dedup_key
|
|
|
|
assert event_dedup_key({"url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci1"}) == (
|
|
"https://earthquake.usgs.gov/earthquakes/eventpage/ci1"
|
|
)
|
|
assert event_dedup_key({"url": " "}) is None
|
|
assert event_dedup_key({}) is None
|
|
|
|
|
|
def test_usgs_feature_keeps_id_and_url():
|
|
from sources import parse_usgs_feature
|
|
|
|
feature = {
|
|
"id": "ci39818991",
|
|
"properties": {
|
|
"title": "M 2.1 - 5 km W of",
|
|
"url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci39818991",
|
|
"place": "5 km W of",
|
|
"mag": 2.1,
|
|
"time": 1_700_000_000_000,
|
|
},
|
|
"geometry": {"coordinates": [-118.5, 34.1, 10.0]},
|
|
}
|
|
event = parse_usgs_feature(feature)
|
|
assert event["url"] == "https://earthquake.usgs.gov/earthquakes/eventpage/ci39818991"
|
|
assert event["raw"]["usgs_id"] == "ci39818991"
|
|
assert event["location_lat"] == 34.1
|
|
assert event["location_lon"] == -118.5
|
|
assert event["source_type"] == "earthquake"
|
|
|
|
|
|
def test_gdelt_uses_doc_api_and_query_param():
|
|
from sources import GDELT_API, gdelt_params
|
|
|
|
assert GDELT_API == "https://api.gdeltproject.org/api/v2/doc/doc"
|
|
params = gdelt_params(query="unrest", max_articles=50)
|
|
assert params["query"] == "unrest"
|
|
assert "search" not in params
|
|
assert params["mode"] == "ArtList"
|
|
assert params["format"] == "json"
|
|
assert int(params["maxrecords"]) == 50
|
|
|
|
|
|
def test_gdelt_default_query_when_empty():
|
|
from sources import gdelt_params
|
|
|
|
params = gdelt_params(query="", max_articles=25)
|
|
assert params["query"]
|
|
assert "unrest" in params["query"].lower() or "cyber" in params["query"].lower()
|
|
|
|
|
|
def test_parse_gdelt_articles_maps_doc_payload():
|
|
from sources import parse_gdelt_articles
|
|
|
|
payload = {
|
|
"articles": [
|
|
{
|
|
"url": "https://example.com/a",
|
|
"title": "Outage",
|
|
"seendate": "20240101T120000Z",
|
|
"domain": "example.com",
|
|
"language": "English",
|
|
"sourcecountry": "US",
|
|
}
|
|
]
|
|
}
|
|
events = parse_gdelt_articles(payload)
|
|
assert len(events) == 1
|
|
assert events[0]["source_type"] == "gdel-t2"
|
|
assert events[0]["url"] == "https://example.com/a"
|
|
assert events[0]["title"] == "Outage"
|
|
|
|
|
|
def test_ingest_event_skips_duplicate_url(monkeypatch):
|
|
"""Second insert with the same url must not hit events_table.insert."""
|
|
from ingestor import ingest_event
|
|
|
|
calls = {"insert": 0, "dedup": 0}
|
|
|
|
class _Result:
|
|
rowcount = 1
|
|
inserted_primary_key = ["evt-1"]
|
|
|
|
class _Session:
|
|
async def execute(self, stmt):
|
|
sql = str(stmt).lower()
|
|
if "event_dedup" in sql or "on conflict" in sql:
|
|
calls["dedup"] += 1
|
|
self_result = _Result()
|
|
if calls["dedup"] > 1:
|
|
self_result.rowcount = 0
|
|
return self_result
|
|
calls["insert"] += 1
|
|
return _Result()
|
|
|
|
async def commit(self):
|
|
return None
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *a):
|
|
return False
|
|
|
|
import ingestor
|
|
|
|
monkeypatch.setattr(ingestor, "async_session", lambda: _Session())
|
|
|
|
msg = {
|
|
"source_type": "earthquake",
|
|
"title": "M 2.1",
|
|
"url": "https://earthquake.usgs.gov/earthquakes/eventpage/ci1",
|
|
"source_timestamp": datetime(2026, 1, 1, tzinfo=timezone.utc).isoformat(),
|
|
}
|
|
|
|
async def run():
|
|
first = await ingest_event(msg)
|
|
second = await ingest_event(msg)
|
|
return first, second
|
|
|
|
first, second = asyncio.run(run())
|
|
assert first is not None
|
|
assert second is None
|
|
assert calls["insert"] == 1
|