"""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