"""The whole pipeline (main.py) offline: blocks are replaced, no API call.""" import json import sys import types from datetime import date, datetime import pytest from damage_engine import main from damage_engine.detectors import (brina, drought, flood, freezing, hail, high_temperature, hot_wind, snow, strong_wind) from damage_engine.report import report, station_report from damage_engine.schemas import NO_EVENT, Analysis DAY = date(2026, 10, 20) @pytest.fixture def out(tmp_path, monkeypatch): """Send every stored file to tmp_path and use one station (566).""" monkeypatch.setattr(report, "REPORTS_DIR", tmp_path / "reports") monkeypatch.setattr(station_report, "STATION_REPORTS_DIR", tmp_path / "station_reports") monkeypatch.setattr(main, "station_ids", lambda: [566]) monkeypatch.setattr(brina, "run", lambda day: [Analysis( block="brina", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"night_min_temp_c": 9.5})]) monkeypatch.setattr(freezing, "run", lambda day: [Analysis( block="freezing", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"day_min_temp_c": 12.0})]) monkeypatch.setattr(strong_wind, "run", lambda day: [Analysis( block="strong_wind", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"max_wind_ms": 6.2})]) monkeypatch.setattr(hot_wind, "run", lambda day: [Analysis( block="hot_wind", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"max_temp_c": 24.0})]) monkeypatch.setattr(high_temperature, "run", lambda day: [Analysis( block="high_temperature", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"max_temp_c": 24.0})]) monkeypatch.setattr(snow, "run", lambda day: [Analysis( block="snow", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"min_cold_c": 6.0})]) monkeypatch.setattr(hail, "run", lambda day: [Analysis( block="hail", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"max_rain_1h_mm": 0.0})]) monkeypatch.setattr(drought, "run", lambda day: [Analysis( block="drought", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, details={"rain_pct_of_normal": 95.0})]) def broken(day): raise ConnectionError("3A API not reachable") monkeypatch.setattr(flood, "run", broken) return tmp_path def test_block_without_code_is_not_ready(monkeypatch): # every block of the diagram has code now: a made-up empty detectors/ file stands in monkeypatch.setitem(sys.modules, "damage_engine.detectors.empty_block", types.ModuleType("damage_engine.detectors.empty_block")) with pytest.raises(main.BlockNotReady, match="no run"): main.load_block("empty_block") with pytest.raises(main.BlockNotReady, match="does not exist"): main.load_block("no_such_block") def test_pipeline_runs_every_block_and_reports_errors(out): summary = main.run_pipeline(DAY) assert list(summary["blocks"]) == [b for b in main.BLOCKS if b != "drought"] assert list(summary["skipped"]) == ["drought"] # monthly: not due on the 20th assert summary["blocks"]["brina"] is None assert "ConnectionError" in summary["blocks"]["flood"] assert summary["blocks"]["freezing"] is None assert summary["blocks"]["strong_wind"] is None assert summary["blocks"]["hot_wind"] is None assert summary["blocks"]["high_temperature"] is None assert summary["blocks"]["snow"] is None assert summary["blocks"]["hail"] is None assert summary["stations"] == {566: None} stored = json.loads((out / "reports" / "flood" / f"{DAY}.json").read_text(encoding="utf-8")) assert "Traceback" in stored["error_detail"] # full detail kept internally only folder = out / "station_reports" / "566" content = json.loads((folder / f"{DAY}.json").read_text(encoding="utf-8")) failed = {e["block"]: e["error"] for e in content["overall"]["errors"]} assert set(failed) == {"flood"} # flood crashed assert _drought_status(content) == "not_run" # skipped, nothing stored assert "error_detail" not in json.dumps(content) # no traceback for the client assert (folder / f"{DAY}.pdf").read_bytes()[:5] == b"%PDF-" def _drought_status(content): return next(b["status"] for b in content["blocks"] if b["block"] == "drought") def test_once_runs_one_block(out): summary = main.run_pipeline(DAY, ["brina"]) assert summary["blocks"] == {"brina": None} summary = main.run_pipeline(DAY, ["drought"]) # --once runs it on any day assert summary["blocks"] == {"drought": None} and summary["skipped"] == {} def test_monthly_block_runs_on_the_first_and_is_shown_afterwards(out): first = date(2026, 11, 1) summary = main.run_pipeline(first) assert summary["blocks"]["drought"] is None and summary["skipped"] == {} later = date(2026, 11, 5) assert "drought" in main.run_pipeline(later)["skipped"] content = json.loads((out / "station_reports" / "566" / f"{later}.json").read_text(encoding="utf-8")) b = next(b for b in content["blocks"] if b["block"] == "drought") assert b["status"] == "no_event" and b["run_day"] == "2026-11-01" and b["stale"] def test_default_run_day_waits_for_the_latest_window(): hour = main.pipeline_hour() # 7: Flood's window ends at 07:00 assert hour == flood.RUN_HOUR assert main.default_run_day(datetime(2026, 10, 21, hour)) == date(2026, 10, 21) assert main.default_run_day(datetime(2026, 10, 21, hour - 1)) == date(2026, 10, 20) # ─────────────────────────── Hail daytime checks ─────────────────────────── JULY = date(2026, 7, 10) def _hail(station, risk): return Analysis(block="hail", run_day=JULY, field_id=f"ST{station}", field_name=f"field {station}", station_id=station, station_name=f"S{station}", status=NO_EVENT, details={"hail_risk": risk}) def test_hail_risk_stations_from_the_morning_run(out): assert main.hail_risk_stations(JULY) is None # no morning run, in season: all assert main.hail_risk_stations(date(2026, 11, 3)) == set() # outside the season: none report.submit("hail", JULY, [_hail(3, True), _hail(6, False), _hail(10, None)]) assert main.hail_risk_stations(JULY) == {3, 10} # unknown risk is checked too def test_hail_daytime_check_runs_only_for_risk_stations_and_keeps_the_morning_run(out, monkeypatch): report.submit("hail", JULY, [_hail(566, True), _hail(3, False)]) calls = [] def fake(day, hour, ids): calls.append((day, hour, ids)) return [Analysis(block="hail_daytime", run_day=day, field_id="ST566", field_name="field 566", station_id=566, station_name="CuneoGiordano", status=NO_EVENT, summary="No hail suspected: no rain in the 6 hours.", details={"check": "daytime", "window_start": "2026-07-10T07:00", "window_end": "2026-07-10T13:00"})] monkeypatch.setattr(hail, "run_daytime", fake) summary = main.run_hail_daytime(JULY, 13) assert calls == [(JULY, 13, {566})] and summary["error"] is None and summary["checked"] == [566] assert [r.get("slot") for r in report.runs("hail")] == [None] # morning run kept assert [r["slot"] for r in report.day_runs("hail_daytime", JULY)] == ["13h"] content = json.loads((out / "station_reports" / "566" / f"{JULY}.json").read_text(encoding="utf-8")) b = next(b for b in content["blocks"] if b["block"] == "hail") assert b["watch_window"] == {"label": "Daytime checks", "start": "2026-07-10T07:00", "end": "2026-07-10T13:00"} assert not any(b["block"] == "hail_daytime" for b in content["blocks"]) def test_hail_daytime_check_skipped_without_risk_and_errors_stored(out, monkeypatch): report.submit("hail", JULY, [_hail(566, False)]) monkeypatch.setattr(hail, "run_daytime", lambda *a: pytest.fail("must not run")) assert main.run_hail_daytime(JULY, 19)["checked"] == [] assert report.day_runs("hail_daytime", JULY) == [] # nothing stored report.submit("hail", JULY, [_hail(566, True)]) def broken(*a): raise ConnectionError("3A API not reachable") monkeypatch.setattr(hail, "run_daytime", broken) assert "ConnectionError" in main.run_hail_daytime(JULY, 19)["error"] [run] = report.day_runs("hail_daytime", JULY) assert run["slot"] == "19h" and "ConnectionError" in run["error"] def test_serve_schedule_pipeline_then_daytime_checks(): assert main.next_job(datetime(2026, 7, 10, 6, 0)) == (datetime(2026, 7, 10, 7, 5), None) assert main.next_job(datetime(2026, 7, 10, 7, 5)) == (datetime(2026, 7, 10, 13, 5), 13) assert main.next_job(datetime(2026, 7, 10, 14, 0)) == (datetime(2026, 7, 10, 19, 5), 19) assert main.next_job(datetime(2026, 7, 10, 19, 30)) == (datetime(2026, 7, 11, 7, 5), None)