diff --git a/work/meshai/config.py b/work/meshai/config.py index 03d42e3..33dd67a 100644 --- a/work/meshai/config.py +++ b/work/meshai/config.py @@ -357,6 +357,7 @@ class NWSConfig(_SourcedFeed): areas: list = field(default_factory=lambda: ["ID"]) severity_min: str = "moderate" user_agent: str = "" + base_url: str = "https://api.weather.gov/alerts/active" @dataclass @@ -364,6 +365,16 @@ class SWPCConfig(_SourcedFeed): """NOAA Space Weather settings.""" enabled: bool = True + # Per-endpoint URLs (poll intervals are fixed in the adapter, not + # config-driven). Defaults are the historical hardcoded endpoints. + endpoints: dict = field( + default_factory=lambda: { + "scales": "https://services.swpc.noaa.gov/products/noaa-scales.json", + "kp": "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json", + "alerts": "https://services.swpc.noaa.gov/products/alerts.json", + "f107": "https://services.swpc.noaa.gov/json/f107_cm_flux.json", + } + ) @dataclass @@ -375,6 +386,7 @@ class DuctingConfig(_SourcedFeed): tick_seconds: int = 10800 # 3 hours latitude: float = 42.56 # Twin Falls area default longitude: float = -114.47 + base_url: str = "https://api.open-meteo.com/v1/gfs" @dataclass @@ -384,6 +396,14 @@ class NICFFiresConfig(_SourcedFeed): enabled: bool = False tick_seconds: int = 600 state: str = "US-ID" + feed_url: str = ( + "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" + "WFIGS_Interagency_Perimeters_Current/FeatureServer/0/query" + ) + points_url: str = ( + "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" + "WFIGS_Incident_Locations_Current/FeatureServer/0/query" + ) @dataclass @@ -394,6 +414,7 @@ class AvalancheConfig(_SourcedFeed): tick_seconds: int = 1800 center_ids: list = field(default_factory=lambda: ["SNFAC"]) season_months: list = field(default_factory=lambda: [12, 1, 2, 3, 4]) + base_url: str = "https://api.avalanche.org/v2/public/products/map-layer" @dataclass @@ -404,6 +425,9 @@ class USGSConfig(_SourcedFeed): tick_seconds: int = 900 # Minimum 15 min per USGS guidelines sites: list = field(default_factory=list) # Site IDs, e.g. ["13090500"] flood_thresholds: dict = field(default_factory=dict) # {site_id: {flow: X, height: Y}} + base_url: str = "https://waterservices.usgs.gov/nwis/iv/" + nwps_base_url: str = "https://api.water.noaa.gov/nwps/v1/gauges" + site_info_url: str = "https://waterservices.usgs.gov/nwis/site/" @dataclass @@ -432,6 +456,7 @@ class TomTomConfig(_SourcedFeed): tick_seconds: int = 300 api_key: str = "" # Supports ${ENV_VAR} corridors: list = field(default_factory=list) # [{name, lat, lon}, ...] + base_url: str = "https://api.tomtom.com/traffic/services/4/flowSegmentData/relative0/10/json" @dataclass @@ -486,6 +511,7 @@ class FIRMSConfig(_SourcedFeed): day_range: int = 1 # 1-10 days of data confidence_min: str = "nominal" # low, nominal, high proximity_km: float = 10.0 # km to match known fire + base_url: str = "https://firms.modaps.eosdis.nasa.gov/api/area/csv" @@ -511,6 +537,7 @@ class SatpassConfig(_SourcedFeed): # Celestrak GP selectors the fetcher pulls (env.tle_fetch): tle_groups: list = field(default_factory=lambda: ["weather", "stations"]) norad_ids: list = field(default_factory=list) + tle_base_url: str = "https://celestrak.org/NORAD/elements/gp.php" # TLE refresh cadence — TLEs update ~daily, so poll every 6h. tle_refresh_seconds: int = 21600 # Predictor pass filters (used by the next task): diff --git a/work/meshai/env/avalanche.py b/work/meshai/env/avalanche.py index ae50cc8..87e0b89 100644 --- a/work/meshai/env/avalanche.py +++ b/work/meshai/env/avalanche.py @@ -16,6 +16,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + + class AvalancheAdapter: """Avalanche.org map layer polling.""" @@ -33,6 +40,7 @@ class AvalancheAdapter: } def __init__(self, config: "AvalancheConfig", coverage: dict = None): + self._base_url = _cfg_str(config, "base_url", self.BASE_URL) self._center_ids = coverage["center_ids"] if coverage is not None else config.center_ids self._tick_interval = config.tick_seconds or 1800 self._season_months = config.season_months or [12, 1, 2, 3, 4] @@ -77,7 +85,7 @@ class AvalancheAdapter: any_error = False for center_id in self._center_ids: - url = f"{self.BASE_URL}/{center_id}" + url = f"{self._base_url}/{center_id}" headers = { "User-Agent": "MeshAI/1.0", diff --git a/work/meshai/env/ducting.py b/work/meshai/env/ducting.py index 0f3a4f4..72418f2 100644 --- a/work/meshai/env/ducting.py +++ b/work/meshai/env/ducting.py @@ -16,6 +16,15 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +DEFAULT_BASE_URL = "https://api.open-meteo.com/v1/gfs" + + +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + # Pressure levels and approximate heights (meters) PRESSURE_LEVELS = { @@ -35,6 +44,7 @@ class DuctingAdapter: else: self._lat = config.latitude self._lon = config.longitude + self._base_url = _cfg_str(config, "base_url", DEFAULT_BASE_URL) self._tick_interval = config.tick_seconds or 10800 # 3 hours self._last_tick = 0.0 self._status = {} @@ -75,7 +85,7 @@ class DuctingAdapter: ] url = ( - f"https://api.open-meteo.com/v1/gfs" + f"{self._base_url}" f"?latitude={self._lat}&longitude={self._lon}" f"&hourly={','.join(hourly_vars)}" f"&forecast_days=1&timezone=auto" diff --git a/work/meshai/env/fires.py b/work/meshai/env/fires.py index 231977b..c21e417 100644 --- a/work/meshai/env/fires.py +++ b/work/meshai/env/fires.py @@ -26,6 +26,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + + class NICFFiresAdapter: """WFIGS ArcGIS fire perimeter + incident-point polling (merged by IrwinID).""" @@ -34,6 +41,8 @@ class NICFFiresAdapter: def __init__(self, config: "NICFFiresConfig", region_anchors: list = None, coverage: dict = None): self._state = config.state + self._feed_url = _cfg_str(config, "feed_url", self.BASE_URL) + self._points_url = _cfg_str(config, "points_url", self.POINTS_URL) self._tick_interval = config.tick_seconds or 600 self._last_tick = 0.0 self._events = [] @@ -145,7 +154,7 @@ class NICFFiresAdapter: perim_ok = False try: params = self._build_query_params() - url = f"{self.BASE_URL}?{urlencode(params)}" + url = f"{self._feed_url}?{urlencode(params)}" req = Request(url, headers=headers) with urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) @@ -197,7 +206,7 @@ class NICFFiresAdapter: points_ok = False try: params = self._build_points_query_params() - url = f"{self.POINTS_URL}?{urlencode(params)}" + url = f"{self._points_url}?{urlencode(params)}" req = Request(url, headers=headers) with urlopen(req, timeout=30) as resp: data = json.loads(resp.read().decode("utf-8")) diff --git a/work/meshai/env/firms.py b/work/meshai/env/firms.py index 46c266b..d049b5f 100644 --- a/work/meshai/env/firms.py +++ b/work/meshai/env/firms.py @@ -15,6 +15,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + + class FIRMSAdapter: """NASA FIRMS satellite fire hotspot polling. @@ -27,6 +34,7 @@ class FIRMSAdapter: BASE_URL = "https://firms.modaps.eosdis.nasa.gov/api/area/csv" def __init__(self, config: "FIRMSConfig", region_anchors: list = None, fires_adapter=None, coverage: dict = None): + self._base_url = _cfg_str(config, "base_url", self.BASE_URL) self._map_key = config.map_key self._source = config.source or "VIIRS_SNPP_NRT" self._bbox = coverage["bbox"] if coverage is not None else config.bbox # [west, south, east, north] @@ -92,7 +100,7 @@ class FIRMSAdapter: # Format bbox as west,south,east,north bbox_str = ",".join(str(c) for c in self._bbox) - url = f"{self.BASE_URL}/{self._map_key}/{self._source}/{bbox_str}/{self._day_range}" + url = f"{self._base_url}/{self._map_key}/{self._source}/{bbox_str}/{self._day_range}" headers = { "User-Agent": "MeshAI/1.0", diff --git a/work/meshai/env/nws.py b/work/meshai/env/nws.py index 935859f..17acb1b 100644 --- a/work/meshai/env/nws.py +++ b/work/meshai/env/nws.py @@ -15,6 +15,15 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +DEFAULT_BASE_URL = "https://api.weather.gov/alerts/active" + + +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + class NWSAlertsAdapter: """NWS Active Alerts -- polls api.weather.gov""" @@ -35,6 +44,7 @@ class NWSAlertsAdapter: self._areas = derived_areas else: self._areas = config.areas or ["ID"] + self._base_url = _cfg_str(config, "base_url", DEFAULT_BASE_URL) self._user_agent = config.user_agent or "(meshai, ops@example.com)" self._severity_min = config.severity_min or "moderate" self._tick_interval = config.tick_seconds or 60 @@ -181,7 +191,7 @@ class NWSAlertsAdapter: True if data changed """ areas = ",".join(self._areas) - url = f"https://api.weather.gov/alerts/active?area={areas}" + url = f"{self._base_url}?area={areas}" headers = { "User-Agent": self._user_agent, diff --git a/work/meshai/env/swpc.py b/work/meshai/env/swpc.py index d349722..7a527a6 100644 --- a/work/meshai/env/swpc.py +++ b/work/meshai/env/swpc.py @@ -15,18 +15,44 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _cfg_dict(config, attr: str, default: dict) -> dict: + """Read a dict config field, falling back to `default` if absent, + empty, or not a real dict (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return dict(value) if isinstance(value, dict) and value else dict(default) + + class SWPCAdapter: """NOAA Space Weather -- multi-endpoint with staggered ticks.""" - # Endpoint definitions: (url, interval_seconds) + # Poll intervals (seconds) per endpoint -- fixed, not config-driven. + INTERVALS = { + "scales": 300, + "kp": 600, + "alerts": 120, + "f107": 86400, + } + + # Default endpoint URLs (overridable via SWPCConfig.endpoints). + DEFAULT_ENDPOINTS = { + "scales": "https://services.swpc.noaa.gov/products/noaa-scales.json", + "kp": "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json", + "alerts": "https://services.swpc.noaa.gov/products/alerts.json", + "f107": "https://services.swpc.noaa.gov/json/f107_cm_flux.json", + } + + # Backward-compat: (url, interval) tuples. No longer used internally + # (see _endpoint_urls / INTERVALS) but kept in case external code still + # references SWPCAdapter.ENDPOINTS. ENDPOINTS = { - "scales": ("https://services.swpc.noaa.gov/products/noaa-scales.json", 300), - "kp": ("https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json", 600), - "alerts": ("https://services.swpc.noaa.gov/products/alerts.json", 120), - "f107": ("https://services.swpc.noaa.gov/json/f107_cm_flux.json", 86400), + "scales": (DEFAULT_ENDPOINTS["scales"], INTERVALS["scales"]), + "kp": (DEFAULT_ENDPOINTS["kp"], INTERVALS["kp"]), + "alerts": (DEFAULT_ENDPOINTS["alerts"], INTERVALS["alerts"]), + "f107": (DEFAULT_ENDPOINTS["f107"], INTERVALS["f107"]), } def __init__(self, config: "SWPCConfig"): + self._endpoint_urls = _cfg_dict(config, "endpoints", self.DEFAULT_ENDPOINTS) self._last_tick = {} # endpoint -> last_tick timestamp self._status = {} self._events = [] @@ -35,7 +61,7 @@ class SWPCAdapter: self._is_loaded = False # Initialize tick times to 0 - for endpoint in self.ENDPOINTS: + for endpoint in self._endpoint_urls: self._last_tick[endpoint] = 0.0 def tick(self) -> bool: @@ -47,8 +73,9 @@ class SWPCAdapter: changed = False now = time.time() - for endpoint, (url, interval) in self.ENDPOINTS.items(): - if now - self._last_tick[endpoint] >= interval: + for endpoint, url in self._endpoint_urls.items(): + interval = self.INTERVALS.get(endpoint, 300) + if now - self._last_tick.get(endpoint, 0.0) >= interval: self._last_tick[endpoint] = now if self._fetch_endpoint(endpoint, url): changed = True diff --git a/work/meshai/env/tle_fetch.py b/work/meshai/env/tle_fetch.py index 4477242..1e26c8a 100644 --- a/work/meshai/env/tle_fetch.py +++ b/work/meshai/env/tle_fetch.py @@ -39,6 +39,13 @@ logger = logging.getLogger(__name__) GP_BASE_URL = "https://celestrak.org/NORAD/elements/gp.php" +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + + def parse_tle_epoch(line1: str) -> str: """Parse the epoch from TLE line 1 into an ISO-8601 UTC string. @@ -113,6 +120,7 @@ class TLEFetchAdapter: def __init__(self, config: "SatpassConfig"): self._config = config + self._tle_base_url = _cfg_str(config, "tle_base_url", GP_BASE_URL) self._last_tick = 0.0 self._last_error: Optional[str] = None self._consecutive_errors = 0 @@ -127,10 +135,10 @@ class TLEFetchAdapter: targets: list[tuple[str, str]] = [] for group in (getattr(self._config, "tle_groups", None) or []): targets.append( - (f"GROUP={group}", f"{GP_BASE_URL}?GROUP={group}&FORMAT=tle")) + (f"GROUP={group}", f"{self._tle_base_url}?GROUP={group}&FORMAT=tle")) for norad in (getattr(self._config, "norad_ids", None) or []): targets.append( - (f"CATNR={norad}", f"{GP_BASE_URL}?CATNR={norad}&FORMAT=tle")) + (f"CATNR={norad}", f"{self._tle_base_url}?CATNR={norad}&FORMAT=tle")) return targets # -- polling -------------------------------------------------------------- diff --git a/work/meshai/env/traffic.py b/work/meshai/env/traffic.py index 784fda8..4ede072 100644 --- a/work/meshai/env/traffic.py +++ b/work/meshai/env/traffic.py @@ -17,12 +17,20 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + + class TomTomTrafficAdapter: """TomTom Traffic Flow Segment Data polling.""" BASE_URL = "https://api.tomtom.com/traffic/services/4/flowSegmentData/relative0/10/json" def __init__(self, config: "TomTomConfig", coverage: dict = None): + self._base_url = _cfg_str(config, "base_url", self.BASE_URL) self._api_key = self._resolve_env(config.api_key or "") if coverage is not None: self._corridors = [ @@ -144,7 +152,7 @@ class TomTomTrafficAdapter: "unit": "MPH", } - url = f"{self.BASE_URL}?{urlencode(params)}" + url = f"{self._base_url}?{urlencode(params)}" headers = { "User-Agent": "MeshAI/1.0", diff --git a/work/meshai/env/usgs.py b/work/meshai/env/usgs.py index 538ee0e..c341f3d 100644 --- a/work/meshai/env/usgs.py +++ b/work/meshai/env/usgs.py @@ -29,13 +29,24 @@ _nwps_cache_time: dict[str, float] = {} NWPS_CACHE_TTL = 86400 * 7 # 7 days +def _cfg_str(config, attr: str, default: str) -> str: + """Read a string config field, falling back to `default` if absent, + empty, or not a real string (e.g. an unconfigured test mock).""" + value = getattr(config, attr, None) + return value if isinstance(value, str) and value else default + + class USGSStreamsAdapter: """USGS instantaneous values for stream gauge readings with NWS flood stages.""" BASE_URL = "https://waterservices.usgs.gov/nwis/iv/" NWPS_BASE_URL = "https://api.water.noaa.gov/nwps/v1/gauges" + SITE_INFO_URL = "https://waterservices.usgs.gov/nwis/site/" def __init__(self, config: "USGSConfig", coverage: dict = None): + self._base_url = _cfg_str(config, "base_url", self.BASE_URL) + self._nwps_base_url = _cfg_str(config, "nwps_base_url", self.NWPS_BASE_URL) + self._site_info_url = _cfg_str(config, "site_info_url", self.SITE_INFO_URL) self._sites = config.sites or [] self._coverage_bbox = coverage["bbox"] if coverage is not None else None self._tick_interval = max(config.tick_seconds or 900, MIN_TICK_SECONDS) @@ -111,7 +122,7 @@ class USGSStreamsAdapter: nws_gauge_id = usgs_site_id # Query NWPS for flood stages - url = f"{self.NWPS_BASE_URL}/{nws_gauge_id}" + url = f"{self._nwps_base_url}/{nws_gauge_id}" headers = { "User-Agent": "MeshAI/1.0 (stream gauge monitoring)", "Accept": "application/json", @@ -167,7 +178,7 @@ class USGSStreamsAdapter: always populated. This is a best-effort lookup. """ # Try USGS site service for metadata including NWS ID - url = f"https://waterservices.usgs.gov/nwis/site/?format=rdb&sites={usgs_site_id}&siteOutput=expanded" + url = f"{self._site_info_url}?format=rdb&sites={usgs_site_id}&siteOutput=expanded" try: req = Request(url, headers={"User-Agent": "MeshAI/1.0"}) @@ -212,7 +223,7 @@ class USGSStreamsAdapter: "sites": site_id, "siteOutput": "expanded", } - url = f"https://waterservices.usgs.gov/nwis/site/?{urlencode(params)}" + url = f"{self._site_info_url}?{urlencode(params)}" try: req = Request(url, headers={"User-Agent": "MeshAI/1.0", "Accept": "application/json"}) @@ -288,7 +299,7 @@ class USGSStreamsAdapter: params = self._build_iv_params(site_ids) - url = f"{self.BASE_URL}?{urlencode(params)}" + url = f"{self._base_url}?{urlencode(params)}" headers = { "User-Agent": "MeshAI/1.0 (stream gauge monitoring)", diff --git a/work/tests/test_configurable_feed_urls.py b/work/tests/test_configurable_feed_urls.py new file mode 100644 index 0000000..7c8ae2f --- /dev/null +++ b/work/tests/test_configurable_feed_urls.py @@ -0,0 +1,362 @@ +"""Tests proving every native environmental adapter's upstream feed URL(s) +are config-driven (mirrors the roads511 pattern). + +For each of the 9 previously-hardcoded adapters, verifies: + (a) with NO config override, the built/fetched URL equals the historical + hardcoded literal (backward compat), and + (b) setting the new config field changes the built URL accordingly. + +HTTP is monkeypatched at the module level (urlopen) -- no network calls. +Where possible, the request URL is captured via a fake urlopen so the test +exercises the real fetch path rather than just constructor plumbing. +""" +from __future__ import annotations + +from urllib.error import URLError + +import pytest + +from meshai.config import ( + AvalancheConfig, + DuctingConfig, + FIRMSConfig, + NICFFiresConfig, + NWSConfig, + SWPCConfig, + SatpassConfig, + TomTomConfig, + USGSConfig, +) +from meshai.env.avalanche import AvalancheAdapter +from meshai.env.ducting import DuctingAdapter +from meshai.env.fires import NICFFiresAdapter +from meshai.env.firms import FIRMSAdapter +from meshai.env.nws import NWSAlertsAdapter +from meshai.env.swpc import SWPCAdapter +from meshai.env.tle_fetch import TLEFetchAdapter +from meshai.env.traffic import TomTomTrafficAdapter +from meshai.env.usgs import USGSStreamsAdapter + + +class _FakeResp: + """Minimal urlopen() context-manager mock returning fixed bytes.""" + + def __init__(self, text: str = "{}"): + self._text = text + + def read(self) -> bytes: + return self._text.encode("utf-8") + + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + +def _capturing_urlopen(captured: list): + """Build a fake urlopen() that records the request's full_url and + returns an empty-but-valid response (parse errors are swallowed by + every adapter's broad except-Exception around parsing).""" + + def _fake(req, timeout=None): + captured.append(req.full_url) + return _FakeResp("{}") + + return _fake + + +# ============================================================ +# 1. nws.py — NWSConfig.base_url +# ============================================================ + +def test_nws_default_url_matches_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.nws.urlopen", _capturing_urlopen(captured)) + adapter = NWSAlertsAdapter(NWSConfig()) + adapter._fetch() + assert captured == ["https://api.weather.gov/alerts/active?area=ID"] + + +def test_nws_override_url_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.nws.urlopen", _capturing_urlopen(captured)) + cfg = NWSConfig(base_url="https://example.test/alerts") + adapter = NWSAlertsAdapter(cfg) + adapter._fetch() + assert captured == ["https://example.test/alerts?area=ID"] + + +# ============================================================ +# 2. swpc.py — SWPCConfig.endpoints (4 endpoints) +# ============================================================ + +def test_swpc_default_endpoints_match_hardcoded(): + adapter = SWPCAdapter(SWPCConfig()) + assert adapter._endpoint_urls == { + "scales": "https://services.swpc.noaa.gov/products/noaa-scales.json", + "kp": "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json", + "alerts": "https://services.swpc.noaa.gov/products/alerts.json", + "f107": "https://services.swpc.noaa.gov/json/f107_cm_flux.json", + } + + +def test_swpc_default_tick_fetches_hardcoded_urls(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.swpc.urlopen", _capturing_urlopen(captured)) + adapter = SWPCAdapter(SWPCConfig()) + adapter.tick() + assert set(captured) == { + "https://services.swpc.noaa.gov/products/noaa-scales.json", + "https://services.swpc.noaa.gov/products/noaa-planetary-k-index.json", + "https://services.swpc.noaa.gov/products/alerts.json", + "https://services.swpc.noaa.gov/json/f107_cm_flux.json", + } + + +def test_swpc_override_endpoint_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.swpc.urlopen", _capturing_urlopen(captured)) + cfg = SWPCConfig(endpoints={"scales": "https://example.test/scales.json"}) + adapter = SWPCAdapter(cfg) + assert adapter._endpoint_urls == {"scales": "https://example.test/scales.json"} + adapter.tick() + assert captured == ["https://example.test/scales.json"] + + +# ============================================================ +# 3. ducting.py — DuctingConfig.base_url +# ============================================================ + +def test_ducting_default_url_matches_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.ducting.urlopen", _capturing_urlopen(captured)) + adapter = DuctingAdapter(DuctingConfig()) + adapter._fetch() + assert len(captured) == 1 + assert captured[0].startswith("https://api.open-meteo.com/v1/gfs?") + + +def test_ducting_override_url_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.ducting.urlopen", _capturing_urlopen(captured)) + cfg = DuctingConfig(base_url="https://example.test/gfs") + adapter = DuctingAdapter(cfg) + adapter._fetch() + assert len(captured) == 1 + assert captured[0].startswith("https://example.test/gfs?") + + +# ============================================================ +# 4. fires.py — NICFFiresConfig.feed_url / .points_url +# ============================================================ + +def test_fires_default_urls_match_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.fires.urlopen", _capturing_urlopen(captured)) + adapter = NICFFiresAdapter(NICFFiresConfig()) + adapter._fetch() + assert len(captured) == 2 + assert captured[0].startswith( + "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" + "WFIGS_Interagency_Perimeters_Current/FeatureServer/0/query?" + ) + assert captured[1].startswith( + "https://services3.arcgis.com/T4QMspbfLg3qTGWY/arcgis/rest/services/" + "WFIGS_Incident_Locations_Current/FeatureServer/0/query?" + ) + + +def test_fires_override_urls_change_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.fires.urlopen", _capturing_urlopen(captured)) + cfg = NICFFiresConfig( + feed_url="https://example.test/perimeters", + points_url="https://example.test/points", + ) + adapter = NICFFiresAdapter(cfg) + adapter._fetch() + assert len(captured) == 2 + assert captured[0].startswith("https://example.test/perimeters?") + assert captured[1].startswith("https://example.test/points?") + + +# ============================================================ +# 5. firms.py — FIRMSConfig.base_url +# ============================================================ + +def test_firms_default_url_matches_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.firms.urlopen", _capturing_urlopen(captured)) + cfg = FIRMSConfig(map_key="test-key", bbox=[-117, 42, -114, 44]) + adapter = FIRMSAdapter(cfg) + adapter._fetch() + assert captured == [ + "https://firms.modaps.eosdis.nasa.gov/api/area/csv/test-key/VIIRS_SNPP_NRT/-117,42,-114,44/1" + ] + + +def test_firms_override_url_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.firms.urlopen", _capturing_urlopen(captured)) + cfg = FIRMSConfig( + map_key="test-key", + bbox=[-117, 42, -114, 44], + base_url="https://example.test/firms", + ) + adapter = FIRMSAdapter(cfg) + adapter._fetch() + assert captured == [ + "https://example.test/firms/test-key/VIIRS_SNPP_NRT/-117,42,-114,44/1" + ] + + +# ============================================================ +# 6. avalanche.py — AvalancheConfig.base_url +# ============================================================ + +def test_avalanche_default_url_matches_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.avalanche.urlopen", _capturing_urlopen(captured)) + cfg = AvalancheConfig(season_months=list(range(1, 13))) # force in-season + adapter = AvalancheAdapter(cfg) + adapter._fetch() + assert captured == ["https://api.avalanche.org/v2/public/products/map-layer/SNFAC"] + + +def test_avalanche_override_url_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.avalanche.urlopen", _capturing_urlopen(captured)) + cfg = AvalancheConfig( + season_months=list(range(1, 13)), + base_url="https://example.test/map-layer", + ) + adapter = AvalancheAdapter(cfg) + adapter._fetch() + assert captured == ["https://example.test/map-layer/SNFAC"] + + +# ============================================================ +# 7. usgs.py — USGSConfig.base_url / .nwps_base_url / .site_info_url +# ============================================================ + +def test_usgs_default_iv_url_matches_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.usgs.urlopen", _capturing_urlopen(captured)) + cfg = USGSConfig(sites=["13090500"]) + adapter = USGSStreamsAdapter(cfg) + adapter._fetch() + assert len(captured) == 1 + assert captured[0].startswith("https://waterservices.usgs.gov/nwis/iv/?") + + +def test_usgs_override_iv_url_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.usgs.urlopen", _capturing_urlopen(captured)) + cfg = USGSConfig(sites=["13090500"], base_url="https://example.test/nwis/iv/") + adapter = USGSStreamsAdapter(cfg) + adapter._fetch() + assert len(captured) == 1 + assert captured[0].startswith("https://example.test/nwis/iv/?") + + +def test_usgs_default_nwps_and_site_info_urls_match_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.usgs.urlopen", _capturing_urlopen(captured)) + adapter = USGSStreamsAdapter(USGSConfig()) + adapter._lookup_nwps_stages("13090500") + # crosswalk (site_info_url) fires first, then the NWPS gauge lookup. + assert len(captured) == 2 + assert captured[0].startswith("https://waterservices.usgs.gov/nwis/site/?") + assert captured[1].startswith("https://api.water.noaa.gov/nwps/v1/gauges/") + + +def test_usgs_override_nwps_and_site_info_urls_change_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.usgs.urlopen", _capturing_urlopen(captured)) + cfg = USGSConfig( + nwps_base_url="https://example.test/nwps", + site_info_url="https://example.test/site/", + ) + adapter = USGSStreamsAdapter(cfg) + # A distinct site id (not "13090500") avoids the module-level + # _nwps_cache/_nwps_cache_time from the default-URL test above short- + # circuiting this call with a cached (stale-URL) result. + adapter._lookup_nwps_stages("09876543") + assert len(captured) == 2 + assert captured[0].startswith("https://example.test/site/?") + assert captured[1].startswith("https://example.test/nwps/") + + +def test_usgs_lookup_site_uses_site_info_url(monkeypatch): + captured: list = [] + + def _fake(req, timeout=None): + captured.append(req.full_url) + raise URLError("no network in test") + + monkeypatch.setattr("meshai.env.usgs.urlopen", _fake) + adapter = USGSStreamsAdapter(USGSConfig()) + adapter.lookup_site("13090500") + assert captured, "lookup_site must attempt a site-info request" + assert captured[0].startswith("https://waterservices.usgs.gov/nwis/site/?") + + +# ============================================================ +# 8. traffic.py — TomTomConfig.base_url +# ============================================================ + +def test_traffic_default_url_matches_hardcoded(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.traffic.urlopen", _capturing_urlopen(captured)) + cfg = TomTomConfig(api_key="test-key") + adapter = TomTomTrafficAdapter(cfg) + adapter._fetch_point("wilderness_cell", 43.5, -115.0, 0.0) + assert len(captured) == 1 + assert captured[0].startswith( + "https://api.tomtom.com/traffic/services/4/flowSegmentData/relative0/10/json?" + ) + + +def test_traffic_override_url_changes_fetch(monkeypatch): + captured: list = [] + monkeypatch.setattr("meshai.env.traffic.urlopen", _capturing_urlopen(captured)) + cfg = TomTomConfig(api_key="test-key", base_url="https://example.test/flow") + adapter = TomTomTrafficAdapter(cfg) + adapter._fetch_point("wilderness_cell", 43.5, -115.0, 0.0) + assert len(captured) == 1 + assert captured[0].startswith("https://example.test/flow?") + + +# ============================================================ +# 9. tle_fetch.py — SatpassConfig.tle_base_url +# ============================================================ + +def test_tle_fetch_default_url_matches_hardcoded(): + cfg = SatpassConfig(tle_groups=["weather"], norad_ids=[25544]) + adapter = TLEFetchAdapter(cfg) + urls = [u for _, u in adapter._targets()] + assert any( + u == "https://celestrak.org/NORAD/elements/gp.php?GROUP=weather&FORMAT=tle" + for u in urls + ) + assert any( + u == "https://celestrak.org/NORAD/elements/gp.php?CATNR=25544&FORMAT=tle" + for u in urls + ) + + +def test_tle_fetch_override_url_changes_targets(): + cfg = SatpassConfig( + tle_groups=["weather"], + norad_ids=[25544], + tle_base_url="https://example.test/gp.php", + ) + adapter = TLEFetchAdapter(cfg) + urls = [u for _, u in adapter._targets()] + assert any( + u == "https://example.test/gp.php?GROUP=weather&FORMAT=tle" for u in urls + ) + assert any( + u == "https://example.test/gp.php?CATNR=25544&FORMAT=tle" for u in urls + )