"""The payload layer: pure builders, span clamping, and the layering guarantee that offline callers (migrate) can use it without dragging in the web stack.""" import os import subprocess import sys import pandas as pd import pytest import climate import views CELL = {"id": "1642_-4223", "center_lat": 47.6087, "center_lon": -122.29377} def test_views_and_migrate_import_without_the_web_stack(): """migrate.py must stay runnable offline: importing the payload layer may not construct the FastAPI app or start the places-index download.""" code = ("import sys; import views, migrate; " "assert 'fastapi' not in sys.modules, 'views/migrate pulled in FastAPI'; " "assert 'app' not in sys.modules, 'views/migrate imported the web app'; " "import places; assert places._load_started is False") backend = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) subprocess.run([sys.executable, "-c", code], cwd=backend, check=True) # ---- cal_span clamping --------------------------------------------------------- def _hist(start, end): df = pd.DataFrame({"date": pd.date_range(start, end, freq="D")}) return df def test_cal_span_defaults_to_months_back_same_day_of_month(): start_ts, end_ts = views.cal_span(_hist("2020-01-01", "2026-06-29"), None, None, 24) assert end_ts == pd.Timestamp("2026-06-29") assert start_ts == pd.Timestamp("2024-06-29") def test_cal_span_clamps_to_the_record(): h = _hist("2025-03-01", "2026-06-29") # record shorter than the 2-year cap start_ts, end_ts = views.cal_span(h, "2020-01-01", None, 24) assert start_ts == pd.Timestamp("2025-03-01") # can't start before the record _, end_ts = views.cal_span(h, None, "2030-01-01", 24) assert end_ts == pd.Timestamp("2026-06-29") # nor end past it def test_cal_span_caps_at_two_years(): start_ts, end_ts = views.cal_span(_hist("2018-01-01", "2026-06-29"), "2018-01-01", "2026-06-29", 24) assert (end_ts - start_ts).days == views.CAL_MAX_SPAN_DAYS - 1 def test_cal_span_never_inverts(): start_ts, end_ts = views.cal_span(_hist("2020-01-01", "2026-06-29"), "2026-06-01", "2021-01-01", 24) assert start_ts == end_ts # ---- builders ------------------------------------------------------------------- def test_build_grade_window_and_shape(history, recent): target = pd.Timestamp.today().normalize() payload = views.build_grade(CELL, target, 14, history, recent, {"cached": True}, "Testville") assert payload["target_date"] == target.date().isoformat() days = [d["date"] for d in payload["recent"]] assert days == sorted(days, reverse=True) # newest first assert payload["climatology"]["tmax"] is not None assert all(d["dsr"] is not None for d in payload["recent"]) def test_build_day_pulls_future_obs_from_recent(history, recent): today = pd.Timestamp.today().normalize() payload = views.build_day(CELL, history, today, "Testville", recent=recent) assert payload["detail"]["date"] == today.date().isoformat() assert payload["detail"]["metrics"]["tmax"]["obs"] is not None def test_build_day_survives_recent_fetch_failure(history, monkeypatch): def boom(cell): raise RuntimeError("upstream down") monkeypatch.setattr(climate, "get_recent_forecast", boom) today = pd.Timestamp.today().normalize() payload = views.build_day(CELL, history, today, "Testville") assert payload["detail"]["metrics"]["tmax"]["obs"] is None # climatology only assert payload["detail"]["metrics"]["tmax"]["ladder"] is not None def test_build_forecast_only_future_days(history, recent): today = pd.Timestamp.today().normalize() payload = views.build_forecast(CELL, 7, history, recent, today, None) days = [d["date"] for d in payload["recent"]] assert days == sorted(days, reverse=True) # furthest-out first assert min(days) > today.date().isoformat() assert payload["forecast"] is True