"""FORECAST_IN_COLLECTOR=true: ping-collector forecasts and notifies; the API serves its stored forecast and falls back to computing when that is stale.""" import asyncio import datetime import json import types import pytest from sqlalchemy import create_engine, text from src import web_api from src.config import Config ROWS = [ { "station_code": "P.1", "horizon_hours": 24, "p_warning": 0.012, "p_danger": 0.001, "predicted_max_level": 1.31, "current_level": 1.23, "as_of": "2026-10-18T13:00:00", "model_version": "hgb-v3+abc", "trained_at": "2026-10-01T03:42:00", "source": "model", "threshold_warning": 3.7, "threshold_danger": 4.2, } ] @pytest.fixture def collector_db(tmp_path, monkeypatch): engine = create_engine(f"sqlite:///{tmp_path / 'db.sqlite'}") with engine.begin() as conn: conn.execute( text( "CREATE TABLE forecast_current (id VARCHAR(16) PRIMARY KEY, " "computed_at TIMESTAMP NOT NULL, model_versions TEXT, payload TEXT NOT NULL)" ) ) scraper = types.SimpleNamespace(db_adapter=types.SimpleNamespace(engine=engine)) monkeypatch.setitem(web_api.app_state, "scraper", scraper) monkeypatch.setattr(Config, "FORECAST_IN_COLLECTOR", True) monkeypatch.setattr(web_api, "FORECAST_CACHE", {}) def store(age_minutes): computed = datetime.datetime.now() - datetime.timedelta(minutes=age_minutes) with engine.begin() as conn: conn.execute(text("DELETE FROM forecast_current")) conn.execute( text( "INSERT INTO forecast_current VALUES ('all', :t, 'hgb-v3+abc', :p)" ), {"t": computed, "p": json.dumps(ROWS)}, ) return store def test_serves_the_collectors_forecast(collector_db, monkeypatch): collector_db(age_minutes=20) def must_not_compute(): raise AssertionError("computed in-process although a fresh forecast is stored") monkeypatch.setattr("src.ml.predict.get_latest_forecasts", must_not_compute) assert asyncio.run(web_api.get_flood_forecasts()) == ROWS # served from the short cache on the next request assert "stored" in web_api.FORECAST_CACHE def test_stale_stored_forecast_falls_back_to_computing(collector_db, monkeypatch): collector_db(age_minutes=Config.FORECAST_MAX_AGE_SECONDS / 60 + 5) computed = [dict(ROWS[0], source="heuristic")] monkeypatch.setattr("src.ml.predict.get_latest_forecasts", lambda: computed) assert asyncio.run(web_api.get_flood_forecasts()) == computed def test_missing_table_falls_back_to_computing(tmp_path, monkeypatch): engine = create_engine(f"sqlite:///{tmp_path / 'empty.sqlite'}") scraper = types.SimpleNamespace(db_adapter=types.SimpleNamespace(engine=engine)) monkeypatch.setitem(web_api.app_state, "scraper", scraper) monkeypatch.setattr(Config, "FORECAST_IN_COLLECTOR", True) monkeypatch.setattr(web_api, "FORECAST_CACHE", {}) monkeypatch.setattr("src.ml.predict.get_latest_forecasts", lambda: ROWS) assert asyncio.run(web_api.get_flood_forecasts()) == ROWS def test_this_app_neither_forecasts_nor_notifies(monkeypatch): published = [] class Publisher: def publish(self, n): published.append(n) return True monkeypatch.setattr(Config, "FORECAST_IN_COLLECTOR", True) monkeypatch.setitem(web_api.app_state, "notify", (Publisher(), object())) monkeypatch.setattr( "src.ml.predict.get_latest_forecasts", lambda: pytest.fail("forecast precompute ran in the API"), ) asyncio.run(web_api._precompute_forecasts()) asyncio.run(web_api._notify_transitions()) assert published == []