qtalker---/backend/tests/test_geomagnetic.py

"""Tests for the NOAA geomagnetic feed.

No test here touches the network. The properties that matter are the
defensive ones: a séance must never fail, block, or crash because a free
third-party service is down, slow, or changed its schema.
"""

import pytest

from app.geomagnetic import (
    STORM_THRESHOLD,
    GeomagneticCache,
    _latest_kp,
    disturbance,
    storm_label,
)


def _payload(*kps):
    """The real shape SWPC's planetary_k_index_1m endpoint serves, verified
    against the live service — a list of objects, not array-of-arrays."""
    return [
        {
            "time_tag": f"2026-07-28T0{i}:00:00",
            "kp_index": int(float(kp)) if kp not in (None, "") else kp,
            "estimated_kp": kp,
            "kp": f"{kp}Z",
        }
        for i, kp in enumerate(kps)
    ]


def _legacy_payload(*kps):
    """The header-row shape some other SWPC endpoints use."""
    rows = [["time_tag", "Kp", "a_running", "station_count"]]
    for i, kp in enumerate(kps):
        rows.append([f"2026-07-28 0{i}:00:00", kp, "4", "8"])
    return rows


class TestLatestKp:
    def test_reads_the_most_recent_row(self):
        assert _latest_kp(_payload("1.00", "2.33", "5.67")) == 5.67

    def test_reads_the_real_live_payload_shape(self):
        # Captured verbatim from the live endpoint.
        live = [
            {"time_tag": "2026-07-27T23:59:00", "kp_index": 3, "estimated_kp": 3.00, "kp": "3Z"},
            {"time_tag": "2026-07-28T05:54:00", "kp_index": 1, "estimated_kp": 1.00, "kp": "1Z"},
        ]
        assert _latest_kp(live) == 1.0

    def test_prefers_the_fractional_estimate_over_the_rounded_integer(self):
        # estimated_kp carries the precision Kp actually has; kp_index is
        # its rounding, so a 4.67 must not be reported as a 4.
        row = [{"time_tag": "t", "kp_index": 4, "estimated_kp": 4.67, "kp": "4Z"}]
        assert _latest_kp(row) == 4.67

    def test_ignores_the_display_string_field(self):
        # "5Z" is a display form with a letter suffix, not a number.
        row = [{"time_tag": "t", "kp": "5Z"}]
        assert _latest_kp(row) is None

    def test_falls_back_to_kp_index_when_the_estimate_is_missing(self):
        row = [{"time_tag": "t", "kp_index": 6}]
        assert _latest_kp(row) == 6.0

    def test_still_handles_the_legacy_header_row_shape(self):
        assert _latest_kp(_legacy_payload("1.00", "3.5")) == 3.5

    def test_finds_the_kp_column_by_header_name(self):
        payload = [
            ["time_tag", "a_running", "Kp"],
            ["2026-07-28 00:00:00", "4", "3.5"],
        ]
        assert _latest_kp(payload) == 3.5

    def test_skips_trailing_rows_with_unusable_values(self):
        payload = _payload("2.00", "4.00")
        payload.append({"time_tag": "2026-07-28T03:00:00", "estimated_kp": None})
        payload.append({"time_tag": "2026-07-28T04:00:00", "estimated_kp": ""})
        assert _latest_kp(payload) == 4.0

    def test_rejects_out_of_scale_values(self):
        # Kp is defined 0-9; anything else is corrupt, not a real storm.
        assert _latest_kp(_payload("42.0")) is None
        assert _latest_kp(_payload("-3.0")) is None

    @pytest.mark.parametrize(
        "junk",
        [None, [], "not a list", {"kp": 5}, [["time_tag", "Kp"]], 42, [[], []]],
    )
    def test_malformed_payloads_return_none_instead_of_raising(self, junk):
        assert _latest_kp(junk) is None

    def test_short_rows_do_not_index_error(self):
        assert _latest_kp([["time_tag", "Kp"], ["2026-07-28"]]) is None

    def test_objects_missing_every_known_field_return_none(self):
        assert _latest_kp([{"time_tag": "t", "something_else": 3}]) is None


class TestStormLabel:
    def test_uses_noaa_severity_bands(self):
        assert storm_label(0.0) == "quiet"
        assert storm_label(4.5) == "unsettled"
        assert storm_label(5.0) == "minor geomagnetic storm"
        assert storm_label(6.0) == "moderate geomagnetic storm"
        assert storm_label(7.0) == "strong geomagnetic storm"
        assert storm_label(9.0) == "severe geomagnetic storm"

    def test_the_official_storm_threshold_is_where_storms_begin(self):
        assert "storm" not in storm_label(STORM_THRESHOLD - 0.1)
        assert "storm" in storm_label(STORM_THRESHOLD)


class TestDisturbance:
    def test_maps_the_scale_onto_zero_to_one(self):
        assert disturbance(0.0) == 0.0
        assert disturbance(9.0) == 1.0
        assert disturbance(4.5) == pytest.approx(0.5)

    def test_clamps_rather_than_extrapolating(self):
        assert disturbance(-5) == 0.0
        assert disturbance(99) == 1.0


class TestCacheResilience:
    @pytest.mark.asyncio
    async def test_cold_cache_with_no_network_reports_none_and_does_not_raise(
        self, monkeypatch
    ):
        # Point at an unroutable host: the lookup must fail silently.
        cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=600)
        assert await cache.get() is None
        assert await cache.reading() is None

    @pytest.mark.asyncio
    async def test_a_failed_refresh_keeps_serving_the_last_real_value(self):
        # An hour-old genuine measurement beats nothing, and geomagnetic
        # conditions don't change fast enough for that to mislead.
        cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=0)
        cache._kp = 6.0
        assert await cache.get() == 6.0

    @pytest.mark.asyncio
    async def test_reading_reports_storm_state(self):
        cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=600)
        cache._kp = 7.33
        cache._fetched_at = float("inf")
        reading = await cache.reading()
        assert reading["kp"] == 7.33
        assert reading["storm"] is True
        assert reading["label"] == "strong geomagnetic storm"
        assert 0.0 <= reading["disturbance"] <= 1.0

    @pytest.mark.asyncio
    async def test_quiet_conditions_are_not_reported_as_a_storm(self):
        cache = GeomagneticCache(url="http://127.0.0.1:9/never", ttl_s=600)
        cache._kp = 2.0
        cache._fetched_at = float("inf")
        reading = await cache.reading()
        assert reading["storm"] is False
        assert reading["label"] == "quiet"

    @pytest.mark.asyncio
    async def test_cached_kp_never_performs_io(self):
        # The synchronous accessor exists precisely so hot paths can read
        # the value without awaiting anything.
        cache = GeomagneticCache(url="http://127.0.0.1:9/never")
        assert cache.cached_kp is None
        cache._kp = 3.0
        assert cache.cached_kp == 3.0