From 95bc50f572d1dca2b10e23b499db9164c49ebe52 Mon Sep 17 00:00:00 2001 From: Ubuntu Date: Tue, 30 Jun 2026 07:05:31 +0000 Subject: [PATCH] feat: add GenericHttpAdapter (kind=generic_http, v0.15.0 PR2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One config-driven adapter class that operators can instantiate many times via distinct config.adapters rows (unique name, kind="generic_http"). Ships class + Pydantic settings schema + unit tests only; no GUI create route, no migration, no seeded row (PR3). Class is dormant until an operator row exists. Key design decisions: - self.name = config.name in __init__ scopes dedup to each instance - domain validated against STREAMS registry at import time - subject_for derives region from event.data["_enriched"]["geocoder"] identically to usgs_quake (no static subject_region setting) - _dig() helper supports dotted paths into nested dicts/lists Also adds: GUI partials (_event_summaries, _event_rows), CONSUMER- INTEGRATION.md §6 subsection, and test_events_feed_frontend.py sample entry — all required by the existing adapter-coverage consistency tests. Co-Authored-By: Claude Sonnet 4.6 --- docs/CONSUMER-INTEGRATION.md | 49 ++ src/central/adapters/generic_http.py | 442 ++++++++++++ .../templates/_event_rows/generic_http.html | 3 + .../_event_summaries/generic_http.html | 2 + tests/test_events_feed_frontend.py | 1 + tests/test_generic_http.py | 666 ++++++++++++++++++ 6 files changed, 1163 insertions(+) create mode 100644 src/central/adapters/generic_http.py create mode 100644 src/central/gui/templates/_event_rows/generic_http.html create mode 100644 src/central/gui/templates/_event_summaries/generic_http.html create mode 100644 tests/test_generic_http.py diff --git a/docs/CONSUMER-INTEGRATION.md b/docs/CONSUMER-INTEGRATION.md index a7e342a..0d75ad7 100644 --- a/docs/CONSUMER-INTEGRATION.md +++ b/docs/CONSUMER-INTEGRATION.md @@ -2023,6 +2023,55 @@ at parameter `00060`, gage height (ft) at `00065`, water temperature (°C) at \ --- +### generic_http — Generic HTTP Source (v0.15.0) + +- **Source:** Any public REST / GeoJSON endpoint; configured per-instance + via `config.adapters.settings.url`. v1 supports public (unauthenticated) + feeds only; auth is a planned follow-up. +- **Stream:** Determined by the `domain` setting, which must match an + existing Central stream domain (e.g. `wx`, `fire`, `quake`). Subject + filter: `central..>`. +- **Subject:** `central..` — `` is derived from + `event.data["_enriched"]["geocoder"]` by the Photon enrichment pipeline + (identical convention to `usgs_quake`). Returns `unknown` when enrichment + has not yet run or found no geo signal. +- **Category:** `.` (default suffix: `alert`, so + e.g. `wx.alert`). +- **Dedup key:** the value resolved by `id_path` within each source item + (required field; coerced to string). +- **Geo:** GeoJSON geometry from `geometry_path` (default `"geometry"`) + when it resolves to a dict; `geo.centroid` is also set for Point + geometries. Falls back to `lat_path` / `lon_path` for non-GeoJSON + responses. +- **Event.data fields:** operator-configured. `title_path` → `data["title"]` + when set; additional fields via `field_mappings` (`source_path → dest_key` + pairs). No guaranteed fixed schema — varies per operator configuration. +- **Settings summary:** + + | Setting | Required | Default | Notes | + |---|---|---|---| + | `url` | yes | — | Polled endpoint (GET, no auth in v1) | + | `domain` | yes | — | Must be an existing stream domain | + | `id_path` | yes | — | Dotted path to stable unique item id | + | `items_path` | no | `features` | Dotted path to the items array | + | `format` | no | `geojson` | `geojson` or `json` (informational) | + | `time_path` | no | `null` | ISO-8601 timestamp path; uses poll time if absent | + | `title_path` | no | `null` | Path → `data["title"]` | + | `geometry_path` | no | `geometry` | GeoJSON geometry path | + | `lat_path` / `lon_path` | no | `null` | Numeric lat/lon paths (non-GeoJSON fallback) | + | `severity_path` | no | `null` | Int severity (0–4) path | + | `category_suffix` | no | `alert` | Appended to domain for Event.category | + | `field_mappings` | no | `[]` | List of `{source_path, dest_key}` | + +- **Cadence:** 300s (5 min) default. +- **Multiple instances:** one `generic_http` class supports many + `config.adapters` rows (each with a distinct instance `name`); dedup is + scoped per-instance. The class is dormant until an operator row is + created — no built-in seeded instance. + +\ +--- + ## 7. Fall-off / removal semantics Central adapters fall into three buckets for handling upstream events that diff --git a/src/central/adapters/generic_http.py b/src/central/adapters/generic_http.py new file mode 100644 index 0000000..98312c5 --- /dev/null +++ b/src/central/adapters/generic_http.py @@ -0,0 +1,442 @@ +"""Generic HTTP / GeoJSON adapter — operator-instantiable, no Python required. + +One class (kind="generic_http") supports many config.adapters rows. Each +row carries a distinct ``name`` (instance identity) and a ``settings`` dict +that drives a separate poll loop. PR3 ships the GUI create route; this PR +ships only the class. + +v1: public feeds only; auth is a follow-up. +""" + +import logging +import sqlite3 +from collections.abc import AsyncIterator +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Literal + +import aiohttp +from pydantic import BaseModel, field_validator +from tenacity import ( + retry, + retry_if_exception_type, + stop_after_attempt, + wait_exponential_jitter, +) + +from central.adapter import SourceAdapter +from central.adapters._subject_helpers import subject_for_region +from central.config_models import AdapterConfig +from central.config_store import ConfigStore +from central.models import Event, Geo +from central.streams import STREAMS + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Derive the valid domain set from the stream registry at import time. +# Subject filters are always "central..>" so we split on "." and +# take index 1. +# --------------------------------------------------------------------------- +_VALID_DOMAINS: frozenset[str] = frozenset( + s.subject_filter.split(".")[1] for s in STREAMS +) + +_DEDUP_DDL = ( + "CREATE TABLE IF NOT EXISTS published_ids (" + "adapter TEXT NOT NULL, event_id TEXT NOT NULL, " + "first_seen TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, " + "last_seen TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, " + "PRIMARY KEY (adapter, event_id))" +) +_DEDUP_IDX = ( + "CREATE INDEX IF NOT EXISTS published_ids_last_seen " + "ON published_ids (last_seen)" +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _dig(obj: Any, path: str) -> Any: + """Walk a nested dict/list by a dotted path string. + + Each segment is tried as a dict key first; if the current node is a + list/tuple the segment is coerced to an integer index. Returns None on + any miss (missing key, out-of-range index, wrong node type). + + Examples:: + + _dig({"a": {"b": 1}}, "a.b") # 1 + _dig({"a": [10, 20]}, "a.1") # 20 + _dig({"a": 1}, "a.b") # None + _dig(None, "anything") # None + """ + parts = path.split(".") + cur = obj + for part in parts: + if cur is None: + return None + if isinstance(cur, dict): + cur = cur.get(part) + elif isinstance(cur, (list, tuple)): + try: + cur = cur[int(part)] + except (ValueError, IndexError): + return None + else: + return None + return cur + + +# --------------------------------------------------------------------------- +# Settings schema +# --------------------------------------------------------------------------- + +class FieldMapping(BaseModel): + """Map one path inside a source item to a key in Event.data. + + Mirrors TomTomFlowSettings / TileCoord so the GUI model_list widget + renders it identically. + """ + source_path: str # dotted path into the source item + dest_key: str # key written into Event.data + + +class GenericHttpSettings(BaseModel): + """Settings schema for GenericHttpAdapter instances.""" + + url: str + """Endpoint to poll (GET, no auth in v1).""" + + format: Literal["geojson", "json"] = "geojson" + """Response format hint (currently informational; both paths use the same + JSON fetch; the geometry_path / lat_path controls shape extraction).""" + + items_path: str = "features" + """Dotted path to the array of items inside the response object.""" + + domain: str + """Event domain; must match an existing NATS stream (e.g. ``wx``, ``fire``, + ``quake``). Determines which JetStream stream the event lands in.""" + + @field_validator("domain") + @classmethod + def _validate_domain(cls, v: str) -> str: + if v not in _VALID_DOMAINS: + valid = ", ".join(sorted(_VALID_DOMAINS)) + raise ValueError( + f"unknown domain '{v}'; must be one of: {valid}" + ) + return v + + category_suffix: str = "alert" + """Appended to domain to form Event.category: ``{domain}.{category_suffix}``.""" + + id_path: str + """Dotted path to a stable, unique identifier within each item (required + for deduplication).""" + + time_path: str | None = None + """Dotted path to an ISO-8601 timestamp string. When None, poll time + (UTC now) is used.""" + + title_path: str | None = None + """Dotted path to a human-readable title → written to ``data['title']``.""" + + geometry_path: str = "geometry" + """(GeoJSON) dotted path to a GeoJSON geometry object within each item. + Ignored when the resolved value is not a dict.""" + + lat_path: str | None = None + """(JSON) dotted path to a numeric latitude. Used only when + geometry_path does not resolve to a geometry dict.""" + + lon_path: str | None = None + """(JSON) dotted path to a numeric longitude. See lat_path.""" + + severity_path: str | None = None + """Dotted path to an integer severity (0-4). When None, severity is + omitted from the event.""" + + field_mappings: list[FieldMapping] = [] + """Extra source-path → dest-key pairs written into Event.data.""" + + +# --------------------------------------------------------------------------- +# Adapter class +# --------------------------------------------------------------------------- + +class GenericHttpAdapter(SourceAdapter): + """Config-driven REST/GeoJSON source adapter. + + A single class that operators can instantiate many times via distinct + ``config.adapters`` rows (each row has a unique ``name`` / instance + identity and ``kind="generic_http"``). No Python required to add a + new source. + """ + + # Class-level identity (kind) — used by discover_adapters() as the + # registry key. Each *instance* overrides self.name = config.name in + # __init__ so dedup and logs are scoped to the instance. + name = "generic_http" + display_name = "Generic HTTP Source" + description = ( + "Config-driven adapter that polls any public REST or GeoJSON endpoint " + "and maps response fields to Central events. Create one row per " + "source; no Python required." + ) + settings_schema = GenericHttpSettings + default_cadence_s = 300 + data_class = "event" + wizard_order = None # not in the setup wizard; created via operator GUI + enrichment_locations = [] + bypass_bbox_filter = False + + def __init__( + self, + config: AdapterConfig, + config_store: ConfigStore, # unused; accepted for signature uniformity + cursor_db_path: Path, + ) -> None: + # Override class-level name with the *instance* name so that the + # inherited dedup helpers (is_published / mark_published / sweep_old_ids) + # scope their SQL queries to this instance, not the generic_http class. + self.name = config.name + self._cursor_db_path = cursor_db_path + self._session: aiohttp.ClientSession | None = None + self._db: sqlite3.Connection | None = None + self._settings = GenericHttpSettings(**config.settings) + + # ------------------------------------------------------------------ + # Lifecycle + # ------------------------------------------------------------------ + + async def startup(self) -> None: + """Initialize HTTP session and dedup tracker.""" + self._session = aiohttp.ClientSession( + headers={"User-Agent": "Central/1.0 (generic_http adapter)"}, + timeout=aiohttp.ClientTimeout(total=30), + ) + self._db = sqlite3.connect(str(self._cursor_db_path)) + self._db.execute(_DEDUP_DDL) + self._db.execute(_DEDUP_IDX) + self._db.commit() + self.sweep_old_ids() + logger.info( + "generic_http adapter started", + extra={"adapter": self.name, "url": self._settings.url}, + ) + + async def shutdown(self) -> None: + """Close HTTP session and database.""" + if self._session: + await self._session.close() + self._session = None + if self._db: + self._db.close() + self._db = None + logger.info("generic_http adapter shut down", extra={"adapter": self.name}) + + # ------------------------------------------------------------------ + # Configuration hot-reload + # ------------------------------------------------------------------ + + async def apply_config(self, new_config: AdapterConfig) -> None: + """Re-parse settings in place (no reconstruction needed).""" + self._settings = GenericHttpSettings(**new_config.settings) + logger.info( + "generic_http config applied", + extra={"adapter": self.name, "url": self._settings.url}, + ) + + # ------------------------------------------------------------------ + # Fetch + # ------------------------------------------------------------------ + + @retry( + stop=stop_after_attempt(3), + wait=wait_exponential_jitter(initial=1, max=15), + retry=retry_if_exception_type((aiohttp.ClientError,)), + reraise=True, + ) + async def _fetch(self, url: str) -> dict[str, Any]: + """GET ``url`` and return parsed JSON. Retried up to 3 times.""" + if not self._session: + raise RuntimeError("Session not initialized") + async with self._session.get(url) as resp: + resp.raise_for_status() + return await resp.json(content_type=None) + + # ------------------------------------------------------------------ + # Poll + # ------------------------------------------------------------------ + + async def poll(self) -> AsyncIterator[Event]: + """Fetch the configured URL and yield new Events.""" + self.sweep_old_ids() + + s = self._settings + try: + raw = await self._fetch(s.url) + except Exception as exc: + logger.error( + "generic_http fetch failed", + extra={"adapter": self.name, "url": s.url, "error": str(exc)}, + ) + raise + + items = _dig(raw, s.items_path) + if not isinstance(items, list): + logger.warning( + "generic_http items_path did not resolve to a list", + extra={ + "adapter": self.name, + "items_path": s.items_path, + "got": type(items).__name__, + }, + ) + return + + new_count = 0 + for item in items: + event = self._item_to_event(item) + if event is None: + continue + if self.is_published(event.id): + continue + yield event + self.mark_published(event.id) + new_count += 1 + + logger.info( + "generic_http poll completed", + extra={"adapter": self.name, "count": new_count}, + ) + + # ------------------------------------------------------------------ + # Item → Event + # ------------------------------------------------------------------ + + def _item_to_event(self, item: Any) -> Event | None: + """Convert a single source item to an Event, or None to skip.""" + s = self._settings + + # --- id (required) --- + raw_id = _dig(item, s.id_path) + if raw_id is None: + logger.warning( + "generic_http item missing id", + extra={"adapter": self.name, "id_path": s.id_path}, + ) + return None + event_id = str(raw_id) + + # --- category --- + category = f"{s.domain}.{s.category_suffix}" + + # --- time --- + if s.time_path: + raw_time = _dig(item, s.time_path) + if raw_time is not None: + try: + # Python 3.11+ fromisoformat handles "Z"; older needs replace. + event_time = datetime.fromisoformat( + str(raw_time).replace("Z", "+00:00") + ) + if event_time.tzinfo is None: + event_time = event_time.replace(tzinfo=timezone.utc) + except (ValueError, TypeError): + logger.warning( + "generic_http could not parse time; using now", + extra={ + "adapter": self.name, + "time_path": s.time_path, + "raw": raw_time, + }, + ) + event_time = datetime.now(timezone.utc) + else: + event_time = datetime.now(timezone.utc) + else: + event_time = datetime.now(timezone.utc) + + # --- geo --- + geo_kwargs: dict[str, Any] = {} + + # Try GeoJSON geometry first. + if s.geometry_path: + geom = _dig(item, s.geometry_path) + if isinstance(geom, dict): + geo_kwargs["geometry"] = geom + # Also set centroid when geometry is a simple Point. + if geom.get("type") == "Point": + coords = geom.get("coordinates") or [] + if len(coords) >= 2: + try: + geo_kwargs["centroid"] = ( + float(coords[0]), + float(coords[1]), + ) + except (TypeError, ValueError): + pass + + # Fall back to explicit lat/lon paths. + if "geometry" not in geo_kwargs and s.lat_path and s.lon_path: + raw_lat = _dig(item, s.lat_path) + raw_lon = _dig(item, s.lon_path) + if raw_lat is not None and raw_lon is not None: + try: + geo_kwargs["centroid"] = (float(raw_lon), float(raw_lat)) + except (TypeError, ValueError): + pass + + geo = Geo(**geo_kwargs) + + # --- data --- + data: dict[str, Any] = {} + if s.title_path: + title = _dig(item, s.title_path) + if title is not None: + data["title"] = title + + for fm in s.field_mappings: + data[fm.dest_key] = _dig(item, fm.source_path) + + # --- severity --- + severity: int | None = None + if s.severity_path: + raw_sev = _dig(item, s.severity_path) + if raw_sev is not None: + try: + severity = int(raw_sev) + except (TypeError, ValueError): + pass + + return Event( + id=event_id, + adapter=self.name, + category=category, + time=event_time, + geo=geo, + data=data, + severity=severity, + ) + + # ------------------------------------------------------------------ + # Subject routing + # ------------------------------------------------------------------ + + def subject_for(self, event: Event) -> str: + """Return the NATS subject for a generic-http event. + + Pattern: ``central..`` + + Region is derived from ``event.data["_enriched"]["geocoder"]`` + identically to usgs_quake — the Photon enrichment pipeline populates + that key at publish time. Returns "unknown" when enrichment hasn't + run yet or found no geo signal. + """ + region = subject_for_region(event.data) + return f"central.{self._settings.domain}.{region}" diff --git a/src/central/gui/templates/_event_rows/generic_http.html b/src/central/gui/templates/_event_rows/generic_http.html new file mode 100644 index 0000000..5d60642 --- /dev/null +++ b/src/central/gui/templates/_event_rows/generic_http.html @@ -0,0 +1,3 @@ +{# Generic HTTP source. Fields from payload->data->data. #} +{% set d = (event.data.get('data') or {}).get('data') or {} %} +{% if d.get('title') is not none %}
Title
{{ d.title }}
{% endif %} diff --git a/src/central/gui/templates/_event_summaries/generic_http.html b/src/central/gui/templates/_event_summaries/generic_http.html new file mode 100644 index 0000000..a11dff4 --- /dev/null +++ b/src/central/gui/templates/_event_summaries/generic_http.html @@ -0,0 +1,2 @@ +{% set d = (event.data.get('data') or {}).get('data') or {} %} +{%- if d.get('title') %}{{ d.title }}{% endif -%} diff --git a/tests/test_events_feed_frontend.py b/tests/test_events_feed_frontend.py index 00c42de..2dfc5ea 100644 --- a/tests/test_events_feed_frontend.py +++ b/tests/test_events_feed_frontend.py @@ -1189,6 +1189,7 @@ _SAMPLE_INNER = { "current_lat_deg": -17.1487, "current_alt_km": 417.4, }, + "generic_http": {"title": "Generic source alert"}, } # Exact expected subjects for the deterministic adapters. swpc_alerts is omitted diff --git a/tests/test_generic_http.py b/tests/test_generic_http.py new file mode 100644 index 0000000..ad32b49 --- /dev/null +++ b/tests/test_generic_http.py @@ -0,0 +1,666 @@ +"""Unit tests for GenericHttpAdapter. + +Pure unit tests: no database mocking required (we use a real sqlite tmp file +for dedup tests, same as test_usgs_quake.py), and HTTP is patched. +""" + +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from pydantic import ValidationError + +from central.adapters.generic_http import ( + FieldMapping, + GenericHttpAdapter, + GenericHttpSettings, + _dig, +) +from central.config_models import AdapterConfig +from central.models import Event, Geo +from central.streams import STREAMS + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _valid_domain() -> str: + """Return a known valid domain (first event-bearing stream).""" + for s in STREAMS: + domain = s.subject_filter.split(".")[1] + if domain != "meta": + return domain + raise RuntimeError("No non-meta stream found in STREAMS") + + +def make_config( + name: str = "test_generic", + domain: str | None = None, + extra_settings: dict | None = None, +) -> AdapterConfig: + domain = domain or _valid_domain() + settings: dict = { + "url": "https://example.com/feed.json", + "domain": domain, + "id_path": "id", + "items_path": "features", + "geometry_path": "geometry", + } + if extra_settings: + settings.update(extra_settings) + return AdapterConfig( + name=name, + kind="generic_http", + enabled=True, + cadence_s=300, + settings=settings, + updated_at=datetime.now(timezone.utc), + ) + + +@pytest.fixture +def temp_db_path(): + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + yield Path(f.name) + + +@pytest.fixture +def mock_config_store(): + return MagicMock() + + +# --------------------------------------------------------------------------- +# _dig +# --------------------------------------------------------------------------- + +class TestDig: + def test_simple_key(self): + assert _dig({"a": 1}, "a") == 1 + + def test_nested_keys(self): + assert _dig({"a": {"b": {"c": 42}}}, "a.b.c") == 42 + + def test_list_index(self): + assert _dig({"a": [10, 20, 30]}, "a.1") == 20 + + def test_list_index_zero(self): + assert _dig({"items": [{"x": 99}]}, "items.0.x") == 99 + + def test_missing_key_returns_none(self): + assert _dig({"a": 1}, "b") is None + + def test_missing_nested_returns_none(self): + assert _dig({"a": {"b": 1}}, "a.c") is None + + def test_out_of_range_index_returns_none(self): + assert _dig({"a": [1, 2]}, "a.5") is None + + def test_none_root_returns_none(self): + assert _dig(None, "a.b") is None + + def test_non_dict_mid_path_returns_none(self): + assert _dig({"a": 42}, "a.b") is None + + def test_empty_list(self): + assert _dig([], "0") is None + + +# --------------------------------------------------------------------------- +# Domain validation +# --------------------------------------------------------------------------- + +class TestDomainValidation: + def test_unknown_domain_raises(self): + with pytest.raises(ValidationError) as exc_info: + GenericHttpSettings( + url="https://example.com/feed", + domain="totally_unknown_domain_xyz", + id_path="id", + ) + err_str = str(exc_info.value) + assert "unknown domain" in err_str + # Should list valid options in the error message + assert "wx" in err_str or "fire" in err_str + + def test_known_domain_ok(self): + s = GenericHttpSettings( + url="https://example.com/feed", + domain=_valid_domain(), + id_path="id", + ) + assert s.domain == _valid_domain() + + def test_wx_domain_valid(self): + s = GenericHttpSettings( + url="https://example.com/feed", + domain="wx", + id_path="id", + ) + assert s.domain == "wx" + + def test_fire_domain_valid(self): + s = GenericHttpSettings( + url="https://example.com/feed", + domain="fire", + id_path="id", + ) + assert s.domain == "fire" + + +# --------------------------------------------------------------------------- +# Category composition +# --------------------------------------------------------------------------- + +class TestCategoryComposition: + @pytest.mark.asyncio + async def test_category_default_suffix(self, temp_db_path, mock_config_store): + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = { + "features": [ + { + "id": "item-1", + "geometry": {"type": "Point", "coordinates": [-116.0, 43.0]}, + } + ] + } + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mock_fetch: + mock_fetch.return_value = geojson + events = [e async for e in adapter.poll()] + + assert len(events) == 1 + assert events[0].category == "wx.alert" + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_category_custom_suffix(self, temp_db_path, mock_config_store): + config = make_config(domain="wx", extra_settings={"category_suffix": "warning"}) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = { + "features": [ + { + "id": "item-wx-warn", + "geometry": {"type": "Point", "coordinates": [-116.0, 43.0]}, + } + ] + } + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = geojson + events = [e async for e in adapter.poll()] + + assert events[0].category == "wx.warning" + await adapter.shutdown() + + +# --------------------------------------------------------------------------- +# GeoJSON path — geometry + centroid +# --------------------------------------------------------------------------- + +GEOJSON_FIXTURE = { + "type": "FeatureCollection", + "features": [ + { + "id": "feat-001", + "type": "Feature", + "properties": { + "title": "Test Alert", + "severity": 2, + "updated": "2025-01-15T12:00:00Z", + }, + "geometry": { + "type": "Point", + "coordinates": [-116.2, 43.7], + }, + }, + { + "id": "feat-002", + "type": "Feature", + "properties": { + "title": "Polygon Alert", + "severity": 3, + "updated": "2025-01-15T13:00:00Z", + }, + "geometry": { + "type": "Polygon", + "coordinates": [[[-116, 43], [-115, 43], [-115, 44], [-116, 44], [-116, 43]]], + }, + }, + ], +} + + +class TestGeoJsonPath: + @pytest.mark.asyncio + async def test_point_geometry_and_centroid(self, temp_db_path, mock_config_store): + config = make_config( + domain="wx", + extra_settings={ + "items_path": "features", + "id_path": "id", + "geometry_path": "geometry", + "title_path": "properties.title", + "severity_path": "properties.severity", + "field_mappings": [ + {"source_path": "properties.title", "dest_key": "title"}, + ], + }, + ) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = GEOJSON_FIXTURE + events = [e async for e in adapter.poll()] + + assert len(events) == 2 + + point_event = next(e for e in events if e.id == "feat-001") + # geo.geometry set + assert point_event.geo.geometry == { + "type": "Point", + "coordinates": [-116.2, 43.7], + } + # centroid extracted from Point + assert point_event.geo.centroid == (-116.2, 43.7) + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_polygon_geometry_no_centroid(self, temp_db_path, mock_config_store): + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = GEOJSON_FIXTURE + events = [e async for e in adapter.poll()] + + poly_event = next(e for e in events if e.id == "feat-002") + # geometry is set + assert poly_event.geo.geometry is not None + assert poly_event.geo.geometry["type"] == "Polygon" + # centroid is NOT set (non-Point geometry) + assert poly_event.geo.centroid is None + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_field_mappings_populate_data(self, temp_db_path, mock_config_store): + config = make_config( + domain="wx", + extra_settings={ + "id_path": "id", + "title_path": "properties.title", + "field_mappings": [ + {"source_path": "properties.severity", "dest_key": "level"}, + {"source_path": "properties.updated", "dest_key": "updated_at"}, + ], + }, + ) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = GEOJSON_FIXTURE + events = [e async for e in adapter.poll()] + + e = next(ev for ev in events if ev.id == "feat-001") + assert e.data["title"] == "Test Alert" + assert e.data["level"] == 2 + assert e.data["updated_at"] == "2025-01-15T12:00:00Z" + + await adapter.shutdown() + + +# --------------------------------------------------------------------------- +# JSON (non-GeoJSON) path — lat_path / lon_path +# --------------------------------------------------------------------------- + +JSON_FIXTURE = { + "alerts": [ + {"uid": "a1", "name": "Alert One", "lat": 43.5, "lon": -116.1, "sev": 1}, + {"uid": "a2", "name": "Alert Two", "lat": 44.0, "lon": -115.5, "sev": 3}, + ] +} + + +class TestJsonLatLonPath: + @pytest.mark.asyncio + async def test_lat_lon_centroid(self, temp_db_path, mock_config_store): + config = make_config( + domain="fire", + extra_settings={ + "items_path": "alerts", + "id_path": "uid", + "geometry_path": "", # disable geometry_path extraction + "lat_path": "lat", + "lon_path": "lon", + "title_path": "name", + "severity_path": "sev", + }, + ) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = JSON_FIXTURE + events = [e async for e in adapter.poll()] + + assert len(events) == 2 + e1 = next(e for e in events if e.id == "a1") + # centroid is (lon, lat) per GeoJSON convention + assert e1.geo.centroid == (-116.1, 43.5) + assert e1.data["title"] == "Alert One" + assert e1.severity == 1 + + await adapter.shutdown() + + +# --------------------------------------------------------------------------- +# Dedup +# --------------------------------------------------------------------------- + +class TestDedup: + @pytest.mark.asyncio + async def test_second_poll_yields_nothing(self, temp_db_path, mock_config_store): + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = { + "features": [ + {"id": "dedup-1", "geometry": None}, + {"id": "dedup-2", "geometry": None}, + ] + } + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = geojson + + first = [e async for e in adapter.poll()] + second = [e async for e in adapter.poll()] + + assert len(first) == 2 + assert len(second) == 0 + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_new_item_in_second_poll_is_yielded( + self, temp_db_path, mock_config_store + ): + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + first_batch = {"features": [{"id": "old-1", "geometry": None}]} + second_batch = { + "features": [ + {"id": "old-1", "geometry": None}, + {"id": "new-2", "geometry": None}, + ] + } + + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = first_batch + [e async for e in adapter.poll()] # consume first + + mf.return_value = second_batch + second = [e async for e in adapter.poll()] + + assert len(second) == 1 + assert second[0].id == "new-2" + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_instance_name_used_for_dedup( + self, temp_db_path, mock_config_store + ): + """Two adapter instances share the same db but scope by instance name.""" + config_a = make_config(name="instance_a", domain="wx") + config_b = make_config(name="instance_b", domain="fire") + + adapter_a = GenericHttpAdapter(config_a, mock_config_store, temp_db_path) + adapter_b = GenericHttpAdapter(config_b, mock_config_store, temp_db_path) + await adapter_a.startup() + await adapter_b.startup() + + # Publish "shared-id" under instance_a + adapter_a.mark_published("shared-id") + + # same id is NOT published under instance_b + assert not adapter_b.is_published("shared-id") + # and IS published under instance_a + assert adapter_a.is_published("shared-id") + + await adapter_a.shutdown() + await adapter_b.shutdown() + + +# --------------------------------------------------------------------------- +# subject_for +# --------------------------------------------------------------------------- + +class TestSubjectFor: + @pytest.mark.asyncio + async def test_subject_no_enrichment(self, temp_db_path, mock_config_store): + """Without enrichment data, subject should end in 'unknown'.""" + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + event = Event( + id="subj-1", + adapter="test_generic", + category="wx.alert", + time=datetime.now(timezone.utc), + geo=Geo(centroid=(-116.0, 43.0)), + data={}, + ) + subject = adapter.subject_for(event) + assert subject == "central.wx.unknown" + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_subject_with_us_enrichment(self, temp_db_path, mock_config_store): + """With US geocoder enrichment, subject should be central..us..""" + config = make_config(domain="fire") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + event = Event( + id="subj-2", + adapter="test_generic", + category="fire.alert", + time=datetime.now(timezone.utc), + geo=Geo(centroid=(-116.0, 43.0)), + data={ + "_enriched": { + "geocoder": { + "country": "United States", + "state": "Idaho", + } + } + }, + ) + subject = adapter.subject_for(event) + assert subject == "central.fire.us.id" + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_subject_format_domain_region(self, temp_db_path, mock_config_store): + """Subject always matches central.. — NOT central...""" + config = make_config(domain="quake", extra_settings={"category_suffix": "event.minor"}) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + event = Event( + id="subj-3", + adapter="test_generic", + category="quake.event.minor", + time=datetime.now(timezone.utc), + geo=Geo(), + data={}, + ) + subject = adapter.subject_for(event) + # Should be central.quake.unknown, NOT central.quake.event.minor.unknown + assert subject == "central.quake.unknown" + assert subject.startswith("central.quake.") + + await adapter.shutdown() + + +# --------------------------------------------------------------------------- +# time_path parsing +# --------------------------------------------------------------------------- + +class TestTimePath: + @pytest.mark.asyncio + async def test_time_path_none_uses_now(self, temp_db_path, mock_config_store): + before = datetime.now(timezone.utc) + config = make_config(domain="wx", extra_settings={"time_path": None}) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = {"features": [{"id": "t1", "geometry": None}]} + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = geojson + events = [e async for e in adapter.poll()] + + after = datetime.now(timezone.utc) + assert len(events) == 1 + assert before <= events[0].time <= after + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_time_path_parsed_iso(self, temp_db_path, mock_config_store): + config = make_config( + domain="wx", + extra_settings={ + "time_path": "properties.updated", + "id_path": "id", + "items_path": "features", + }, + ) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = { + "features": [ + { + "id": "t2", + "geometry": None, + "properties": {"updated": "2025-06-15T08:30:00Z"}, + } + ] + } + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = geojson + events = [e async for e in adapter.poll()] + + assert len(events) == 1 + assert events[0].time == datetime(2025, 6, 15, 8, 30, 0, tzinfo=timezone.utc) + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_time_path_missing_value_uses_now( + self, temp_db_path, mock_config_store + ): + """When time_path is set but the value is absent, fall back to now.""" + before = datetime.now(timezone.utc) + config = make_config( + domain="wx", + extra_settings={"time_path": "properties.ts"}, + ) + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = {"features": [{"id": "t3", "geometry": None, "properties": {}}]} + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = geojson + events = [e async for e in adapter.poll()] + + after = datetime.now(timezone.utc) + assert len(events) == 1 + assert before <= events[0].time <= after + + await adapter.shutdown() + + +# --------------------------------------------------------------------------- +# apply_config +# --------------------------------------------------------------------------- + +class TestApplyConfig: + @pytest.mark.asyncio + async def test_apply_config_updates_url(self, temp_db_path, mock_config_store): + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + assert adapter._settings.url == "https://example.com/feed.json" + + new_config = make_config( + domain="wx", extra_settings={"url": "https://other.example.com/data.json"} + ) + await adapter.apply_config(new_config) + + assert adapter._settings.url == "https://other.example.com/data.json" + + await adapter.shutdown() + + @pytest.mark.asyncio + async def test_apply_config_updates_domain(self, temp_db_path, mock_config_store): + config = make_config(domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + new_config = make_config(domain="fire") + await adapter.apply_config(new_config) + + assert adapter._settings.domain == "fire" + + await adapter.shutdown() + + +# --------------------------------------------------------------------------- +# Instance name scoping +# --------------------------------------------------------------------------- + +class TestInstanceName: + def test_instance_name_is_config_name(self, temp_db_path, mock_config_store): + """adapter.name must be the instance name, not the class name.""" + config = make_config(name="my_noaa_alerts", domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + assert adapter.name == "my_noaa_alerts" + assert GenericHttpAdapter.name == "generic_http" + + @pytest.mark.asyncio + async def test_event_adapter_field_is_instance_name( + self, temp_db_path, mock_config_store + ): + config = make_config(name="my_source", domain="wx") + adapter = GenericHttpAdapter(config, mock_config_store, temp_db_path) + await adapter.startup() + + geojson = {"features": [{"id": "inst-1", "geometry": None}]} + with patch.object(adapter, "_fetch", new_callable=AsyncMock) as mf: + mf.return_value = geojson + events = [e async for e in adapter.poll()] + + assert len(events) == 1 + assert events[0].adapter == "my_source" + + await adapter.shutdown()