mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-11-04 03:52:34 +00:00
agessaman
5601a126dd
This brings the weather slices of my long-running refactor branch to dev ahead of converging wx and gwx, so the convergence PRs build on one copy of the code instead of rewriting it twice. - `modules/weather_common.py` (WeatherCommandMixin): code wx and gwx shared by copy (MQTT and WXSIM source lookup, multi-day sending, high/low formatting, the forecast-option parser, the Open-Meteo model loader). - `modules/nws_alerts.py`: one NWS alert entry parser for wx and the weather service, keeping each caller's own Special Weather Statement rules. - `modules/http_retry.py`: one retrying session builder for the weather HTTP clients. - wx: get_noaa_weather, the hourly, tomorrow and multi-day formatters, _add_period_details and the extract_* readers broken up or made table-driven; NOAA fetches go through `_noaa_fetch`. - gwx: one fetch path; `_open_meteo_current` and `_open_meteo_daily_tail` split out of get_open_meteo_weather. - solarforecast: `_format_forecast` built from one timestamp parser and one day builder (418 → 155 lines). - aurora, rain, wx, gwx: companion and bot location lookups through `modules.location` (which gains optional logging arguments). The changes outside these files that the refactor branch made (the base command's enabled_attr hook, the service base's run_periodic and _cancel_tasks) stay out: aurora and rain keep their can_execute overrides and the weather service its own loops, exactly as on dev. Tests: characterization goldens over real captured NOAA, NWS, Open-Meteo and forecast.solar responses (tests/characterization, fixtures in tests/fixtures), 50 tests and several hundred cases, all of which also pass against dev's unmodified weather code. Each refactor step was additionally checked on the branch with an old-vs-new differential run (tens of thousands of generated inputs, including log calls and exceptions) and reviewed by codex.
107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
"""Characterization: how wx and the weather service turn an NWS alerts Atom feed into alert dicts.
|
|
|
|
The two parsers share most of their logic but classify Special Weather
|
|
Statements differently; both behaviors are pinned here. Fixtures are a real
|
|
feed sample (tests/fixtures/nws/alerts_real.atom) and synthetic entries that
|
|
reach each classifier branch (alerts_synthetic.atom).
|
|
"""
|
|
|
|
import configparser
|
|
import re
|
|
import xml.dom.minidom
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
|
|
from tests.characterization.golden_util import assert_golden
|
|
|
|
FIXTURES = Path(__file__).resolve().parents[1] / "fixtures" / "nws"
|
|
FEEDS = ["alerts_real", "alerts_synthetic"]
|
|
|
|
|
|
def _bot(logger, config):
|
|
bot = Mock()
|
|
bot.logger = logger
|
|
bot.config = config
|
|
bot.db_manager = Mock()
|
|
bot.db_manager.get_cached_geocoding = Mock(return_value=(None, None))
|
|
bot.command_manager = Mock()
|
|
bot.command_manager.send_channel_message = AsyncMock()
|
|
bot.translator.translate.side_effect = lambda key, **kwargs: f"<{key}{sorted(kwargs.items()) or ''}>"
|
|
return bot
|
|
|
|
|
|
def _wx(logger):
|
|
from modules.commands.wx_command import WxCommand
|
|
|
|
config = configparser.ConfigParser()
|
|
config.add_section("Weather")
|
|
config.set("Weather", "weather_provider", "noaa")
|
|
config.add_section("Wx_Command")
|
|
cmd = WxCommand(_bot(logger, config))
|
|
return cmd
|
|
|
|
|
|
def _service(logger):
|
|
from modules.service_plugins.weather_service import WeatherService
|
|
|
|
config = configparser.ConfigParser()
|
|
config.add_section("Weather")
|
|
config.add_section("Weather_Service")
|
|
config.set("Weather_Service", "my_position_lat", "47.6")
|
|
config.set("Weather_Service", "my_position_lon", "-122.3")
|
|
return WeatherService(_bot(logger, config))
|
|
|
|
|
|
def _response(text):
|
|
response = Mock()
|
|
response.ok = True
|
|
response.status_code = 200
|
|
response.text = text
|
|
return response
|
|
|
|
|
|
def _single_entry_feeds(text):
|
|
head = text[: text.index("<entry>")]
|
|
return [head + e + "\n</feed>\n" for e in re.findall(r"<entry>.*?</entry>", text, re.S)]
|
|
|
|
|
|
def _wx_alerts(cmd, text, full):
|
|
cmd.noaa_session = Mock()
|
|
cmd.noaa_session.get = Mock(return_value=_response(text))
|
|
cmd._nws_alerts_available = None
|
|
return cmd.get_weather_alerts_noaa(47.6, -122.3, return_full_data=full)
|
|
|
|
|
|
def _summarize(result):
|
|
if isinstance(result, tuple) and result and isinstance(result[0], list):
|
|
alerts, count = result
|
|
return {"count": count, "alerts": alerts, "key_order": [list(a) for a in alerts]}
|
|
return result
|
|
|
|
|
|
@pytest.mark.parametrize("feed", FEEDS)
|
|
def test_wx_alert_feed(feed, mock_logger):
|
|
text = (FIXTURES / f"{feed}.atom").read_text(encoding="utf-8")
|
|
cmd = _wx(mock_logger)
|
|
data = {
|
|
"full": _summarize(_wx_alerts(cmd, text, True)),
|
|
"compact": _wx_alerts(cmd, text, False),
|
|
"per_entry": [_summarize(_wx_alerts(cmd, t, True)) for t in _single_entry_feeds(text)],
|
|
"warnings": [str(c) for c in mock_logger.warning.call_args_list],
|
|
"errors": [str(c) for c in mock_logger.error.call_args_list],
|
|
}
|
|
assert_golden(f"nws_alerts_wx_{feed}", data)
|
|
|
|
|
|
@pytest.mark.parametrize("feed", FEEDS)
|
|
def test_service_alert_entries(feed, mock_logger):
|
|
text = (FIXTURES / f"{feed}.atom").read_text(encoding="utf-8")
|
|
service = _service(mock_logger)
|
|
parsed = []
|
|
for i, entry in enumerate(xml.dom.minidom.parseString(text).getElementsByTagName("entry")):
|
|
alert = service._parse_alert_entry(entry, f"id-{i}")
|
|
parsed.append({"alert": alert, "key_order": list(alert) if alert else None})
|
|
assert_golden(f"nws_alerts_service_{feed}", parsed)
|