prop/test/microwaveprop/weather/hrrr_client_test.exs
Graham McIntire 71e8f53142
fix: stabilize flaky tests + silence PromEx Oban poller in test
- HrrrClient idx-cache test only invalidated the surface idx URL, but
  fetch_profile also fetches a pressure idx. Previous runs' state for
  the pressure key decided whether the counter landed at 1 (prior run
  cached it, pass) or 2 (cold, fail). Invalidate both + assert 2 total
  fetches to reflect the actual code path.
- CsvImportTest deadlocked against other async DataCase tests when
  inline Oban child jobs upserted iemre_observations/terrain_profiles
  with a shared conflict target. Flip to async: false — same fix as
  ContactWeatherEnqueueWorkerTest earlier this session.
- PromEx.Plugins.Oban runs a 5s telemetry_poller that queries the DB,
  but its poller PID has no sandbox connection in test and crashed
  with DBConnection.OwnershipError on every tick, spamming the log.
  Gate the plugin on a config flag and skip it in config/test.exs;
  prod behaviour unchanged.
2026-04-21 13:49:07 -05:00

558 lines
18 KiB
Elixir
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

defmodule Microwaveprop.Weather.HrrrClientTest do
use ExUnit.Case, async: true
alias Microwaveprop.Weather.HrrrClient
describe "nearest_hrrr_hour/1" do
test "rounds down when within first 30 minutes" do
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:15:00Z]) ==
~U[2026-03-28 18:00:00Z]
end
test "rounds up when past 30 minutes" do
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:45:00Z]) ==
~U[2026-03-28 19:00:00Z]
end
test "stays same when exactly on the hour" do
assert HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:00:00Z]) ==
~U[2026-03-28 18:00:00Z]
end
test "rounds at exactly 30 minutes" do
result = HrrrClient.nearest_hrrr_hour(~U[2026-03-28 18:30:00Z])
assert result in [~U[2026-03-28 18:00:00Z], ~U[2026-03-28 19:00:00Z]]
end
end
describe "hrrr_url/3" do
test "builds surface product URL" do
url = HrrrClient.hrrr_url(~D[2026-03-28], 18, :surface)
assert url ==
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfsfcf00.grib2"
end
test "builds pressure product URL" do
url = HrrrClient.hrrr_url(~D[2026-03-28], 6, :pressure)
assert url ==
"https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t06z.wrfprsf00.grib2"
end
test "pads single-digit hours" do
url = HrrrClient.hrrr_url(~D[2026-03-28], 3, :surface)
assert String.contains?(url, "hrrr.t03z")
end
end
describe "parse_idx/1" do
@sample_idx """
1:0:d=2026032818:TMP:2 m above ground:anl:
2:1234567:d=2026032818:DPT:2 m above ground:anl:
3:2345678:d=2026032818:PRES:surface:anl:
4:3456789:d=2026032818:HPBL:surface:anl:
5:4567890:d=2026032818:PWAT:entire atmosphere (considered as a single layer):anl:
6:5678901:d=2026032818:TMP:1000 mb:anl:
"""
test "parses idx into structured list" do
entries = HrrrClient.parse_idx(@sample_idx)
assert length(entries) == 6
first = hd(entries)
assert first.msg == 1
assert first.offset == 0
assert first.var == "TMP"
assert first.level == "2 m above ground"
end
test "correctly extracts offsets" do
entries = HrrrClient.parse_idx(@sample_idx)
offsets = Enum.map(entries, & &1.offset)
assert offsets == [0, 1_234_567, 2_345_678, 3_456_789, 4_567_890, 5_678_901]
end
test "handles empty input" do
assert HrrrClient.parse_idx("") == []
end
test "skips malformed lines instead of raising" do
# NOAA S3 occasionally serves an HTML error page as the idx body
# (e.g. during brief 503 spikes). The old parser would raise
# ArgumentError on String.to_integer of a non-numeric token,
# which killed the whole PropagationGridWorker chain step.
# The tolerant parser returns the well-formed lines and drops
# the garbage.
input = """
<html><body>Service Unavailable</body></html>
1:0:d=2026032818:TMP:2 m above ground:anl:
bogus:header:line:with:no:numbers
2:1234567:d=2026032818:DPT:2 m above ground:anl:
"""
entries = HrrrClient.parse_idx(input)
assert length(entries) == 2
assert Enum.map(entries, & &1.msg) == [1, 2]
assert Enum.map(entries, & &1.var) == ["TMP", "DPT"]
end
end
describe "byte_ranges_for_messages/2" do
test "computes byte ranges from idx entries" do
entries = [
%{msg: 1, offset: 0, var: "TMP", level: "2 m above ground"},
%{msg: 2, offset: 1000, var: "DPT", level: "2 m above ground"},
%{msg: 3, offset: 2000, var: "PRES", level: "surface"},
%{msg: 4, offset: 3000, var: "HPBL", level: "surface"}
]
wanted = [
%{var: "TMP", level: "2 m above ground"},
%{var: "PRES", level: "surface"}
]
ranges = HrrrClient.byte_ranges_for_messages(entries, wanted)
assert length(ranges) == 2
assert {0, 999} in ranges
assert {2000, 2999} in ranges
end
test "handles last message in file" do
entries = [
%{msg: 1, offset: 0, var: "TMP", level: "2 m above ground"},
%{msg: 2, offset: 1000, var: "DPT", level: "2 m above ground"}
]
wanted = [%{var: "DPT", level: "2 m above ground"}]
ranges = HrrrClient.byte_ranges_for_messages(entries, wanted)
# Last message - uses a large end offset
assert length(ranges) == 1
[{start, _end}] = ranges
assert start == 1000
end
end
describe "fetch_profile/3" do
test "succeeds when ranges are downloaded individually" do
grib_data = File.read!("test/fixtures/grib2/hrrr_tmp_2m.grib2")
# Idx with multiple surface variables to trigger multi-range download
grib_size = byte_size(grib_data)
idx_text = """
1:0:d=2026032818:TMP:2 m above ground:anl:
2:#{grib_size}:d=2026032818:DPT:2 m above ground:anl:
3:#{grib_size * 2}:d=2026032818:PRES:surface:anl:
"""
Req.Test.stub(HrrrClient, fn conn ->
if String.ends_with?(conn.request_path, ".idx") do
Plug.Conn.send_resp(conn, 200, idx_text)
else
[range] = Plug.Conn.get_req_header(conn, "range")
if String.contains?(range, ",") do
# Multi-range: S3 ignores Range header, returns full file
Plug.Conn.send_resp(conn, 200, "full file")
else
# Single range: S3 supports this fine
Plug.Conn.send_resp(conn, 206, grib_data)
end
end
end)
assert {:ok, profile} = HrrrClient.fetch_profile(32.90, -97.04, ~U[2026-03-28 18:00:00Z])
assert is_float(profile.surface_temp_c)
assert is_struct(profile.run_time, DateTime)
end
test "caches idx responses across calls to the same HRRR run" do
# `.idx` files are static once a model run is published. The 114s
# `hrrr_fetch_grid` span in prod spends ~10s re-fetching the same idx
# URL across forecast hours / products. Cache eliminates that per-run.
grib_data = File.read!("test/fixtures/grib2/hrrr_tmp_2m.grib2")
grib_size = byte_size(grib_data)
idx_text = """
1:0:d=2026032818:TMP:2 m above ground:anl:
2:#{grib_size}:d=2026032818:DPT:2 m above ground:anl:
3:#{grib_size * 2}:d=2026032818:PRES:surface:anl:
"""
Microwaveprop.Cache.invalidate(
{:hrrr_idx, "https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfsfcf00.grib2.idx"}
)
# `do_fetch_profile` fetches TWO idx URLs (surface + pressure). Also drop
# the pressure cache so the test starts from a deterministic state —
# without this, whether this assertion passes depends on seed ordering.
Microwaveprop.Cache.invalidate(
{:hrrr_idx, "https://noaa-hrrr-bdp-pds.s3.amazonaws.com/hrrr.20260328/conus/hrrr.t18z.wrfprsf00.grib2.idx"}
)
{:ok, counter} = Agent.start_link(fn -> %{idx: 0, grib: 0} end)
Req.Test.stub(HrrrClient, fn conn ->
if String.ends_with?(conn.request_path, ".idx") do
Agent.update(counter, &Map.update!(&1, :idx, fn n -> n + 1 end))
Plug.Conn.send_resp(conn, 200, idx_text)
else
Agent.update(counter, &Map.update!(&1, :grib, fn n -> n + 1 end))
[range] = Plug.Conn.get_req_header(conn, "range")
if String.contains?(range, ",") do
Plug.Conn.send_resp(conn, 200, "full file")
else
Plug.Conn.send_resp(conn, 206, grib_data)
end
end
end)
valid_time = ~U[2026-03-28 18:00:00Z]
assert {:ok, _} = HrrrClient.fetch_profile(32.90, -97.04, valid_time)
assert {:ok, _} = HrrrClient.fetch_profile(32.90, -97.04, valid_time)
%{idx: idx_count, grib: grib_count} = Agent.get(counter, & &1)
# fetch_profile pulls 2 idx URLs (surface + pressure) on the first call;
# the second call must be a full cache hit, so the total stays at 2.
assert idx_count == 2, "idx should be cached — expected 2 total fetches, got #{idx_count}"
assert grib_count > 0, "grib fetches should still happen"
end
end
describe "build_profile/1" do
test "assembles profile from parsed data" do
parsed = %{
"TMP:1000 mb" => 298.0,
"DPT:1000 mb" => 291.0,
"HGT:1000 mb" => 110.0,
"TMP:975 mb" => 296.0,
"DPT:975 mb" => 289.0,
"HGT:975 mb" => 350.0,
"TMP:950 mb" => 294.0,
"DPT:950 mb" => 287.0,
"HGT:950 mb" => 590.0,
"TMP:2 m above ground" => 299.0,
"DPT:2 m above ground" => 292.0,
"PRES:surface" => 101_350.0,
"HPBL:surface" => 1500.0,
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
}
result = HrrrClient.build_profile(parsed)
assert result.surface_temp_c == 299.0 - 273.15
assert result.surface_dewpoint_c == 292.0 - 273.15
assert_in_delta result.surface_pressure_mb, 1013.5, 0.1
assert result.hpbl_m == 1500.0
assert result.pwat_mm == 25.0
assert length(result.profile) == 3
first_level = hd(result.profile)
assert first_level["pres"] == 1000.0
assert first_level["tmpc"] == 298.0 - 273.15
assert first_level["dwpc"] == 291.0 - 273.15
assert first_level["hght"] == 110.0
end
test "includes wind, cloud cover, and precip fields" do
parsed = %{
"TMP:2 m above ground" => 299.0,
"DPT:2 m above ground" => 292.0,
"PRES:surface" => 101_350.0,
"HPBL:surface" => 1500.0,
"PWAT:entire atmosphere (considered as a single layer)" => 25.0,
"UGRD:10 m above ground" => 3.5,
"VGRD:10 m above ground" => -2.1,
"TCDC:entire atmosphere" => 75.0,
"APCP:surface" => 1.2
}
result = HrrrClient.build_profile(parsed)
assert result.wind_u == 3.5
assert result.wind_v == -2.1
assert result.cloud_cover_pct == 75.0
assert result.precip_mm == 1.2
end
test "wind, cloud, and precip fields are nil when missing" do
parsed = %{
"TMP:2 m above ground" => 299.0,
"DPT:2 m above ground" => 292.0,
"PRES:surface" => 101_350.0,
"HPBL:surface" => 1500.0,
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
}
result = HrrrClient.build_profile(parsed)
assert result.wind_u == nil
assert result.wind_v == nil
assert result.cloud_cover_pct == nil
assert result.precip_mm == nil
end
test "skips pressure levels with missing data" do
parsed = %{
"TMP:1000 mb" => 298.0,
"DPT:1000 mb" => 291.0,
"HGT:1000 mb" => 110.0,
# 975 mb missing HGT
"TMP:975 mb" => 296.0,
"DPT:975 mb" => 289.0,
"TMP:2 m above ground" => 299.0,
"DPT:2 m above ground" => 292.0,
"PRES:surface" => 101_350.0,
"HPBL:surface" => 1500.0,
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
}
result = HrrrClient.build_profile(parsed)
assert length(result.profile) == 1
end
test "includes upper-air levels above 700 mb for the skew-T plot" do
parsed = %{
"TMP:700 mb" => 275.0,
"DPT:700 mb" => 265.0,
"HGT:700 mb" => 3100.0,
"TMP:500 mb" => 253.0,
"DPT:500 mb" => 238.0,
"HGT:500 mb" => 5800.0,
"TMP:300 mb" => 228.0,
"DPT:300 mb" => 200.0,
"HGT:300 mb" => 9400.0,
"TMP:100 mb" => 205.0,
"DPT:100 mb" => 180.0,
"HGT:100 mb" => 16_300.0,
"TMP:2 m above ground" => 299.0,
"DPT:2 m above ground" => 292.0,
"PRES:surface" => 101_350.0,
"HPBL:surface" => 1500.0,
"PWAT:entire atmosphere (considered as a single layer)" => 25.0
}
result = HrrrClient.build_profile(parsed)
pressures = Enum.map(result.profile, & &1["pres"])
assert 500.0 in pressures
assert 300.0 in pressures
assert 100.0 in pressures
level_500 = Enum.find(result.profile, &(&1["pres"] == 500.0))
assert_in_delta level_500["tmpc"], 253.0 - 273.15, 0.001
assert_in_delta level_500["dwpc"], 238.0 - 273.15, 0.001
assert level_500["hght"] == 5800.0
end
end
describe "pressure_messages/1" do
test ":grid returns only the near-surface levels needed for scoring" do
messages = HrrrClient.pressure_messages(:grid)
# 13 levels × 3 vars (TMP/DPT/HGT)
assert length(messages) == 39
levels =
messages
|> Enum.map(fn m ->
[num_str, _] = String.split(m.level, " ")
String.to_integer(num_str)
end)
|> Enum.uniq()
|> Enum.sort()
# Narrow list: 1000 → 700 mb, every 25 mb
assert Enum.max(levels) == 1000
assert Enum.min(levels) == 700
assert 700 in levels
assert 900 in levels
# Upper-air levels must be absent — they're the whole reason the grid
# worker OOMs when this function returns the full profile list.
refute 500 in levels
refute 300 in levels
refute 100 in levels
end
test ":profile returns full troposphere + lower stratosphere for the skew-T" do
messages = HrrrClient.pressure_messages(:profile)
# 25 levels × 3 vars
assert length(messages) == 75
levels =
messages
|> Enum.map(fn m ->
[num_str, _] = String.split(m.level, " ")
String.to_integer(num_str)
end)
|> Enum.uniq()
|> Enum.sort()
assert Enum.max(levels) == 1000
assert Enum.min(levels) == 100
assert 500 in levels
assert 300 in levels
assert 100 in levels
end
test ":grid and :profile both request TMP, DPT and HGT at every level" do
for variant <- [:grid, :profile] do
messages = HrrrClient.pressure_messages(variant)
vars = messages |> Enum.map(& &1.var) |> Enum.uniq() |> Enum.sort()
assert vars == ["DPT", "HGT", "TMP"]
end
end
end
describe "surface_messages/0" do
test "returns list of surface message descriptors" do
messages = HrrrClient.surface_messages()
assert is_list(messages)
assert length(messages) == 9
vars = Enum.map(messages, & &1.var)
assert "TMP" in vars
assert "DPT" in vars
assert "PRES" in vars
assert "HPBL" in vars
assert "PWAT" in vars
assert "UGRD" in vars
assert "VGRD" in vars
assert "TCDC" in vars
assert "APCP" in vars
end
test "includes wind messages at 10 m above ground" do
messages = HrrrClient.surface_messages()
ugrd = Enum.find(messages, &(&1.var == "UGRD"))
assert ugrd.level == "10 m above ground"
vgrd = Enum.find(messages, &(&1.var == "VGRD"))
assert vgrd.level == "10 m above ground"
end
test "includes cloud cover for entire atmosphere" do
messages = HrrrClient.surface_messages()
tcdc = Enum.find(messages, &(&1.var == "TCDC"))
assert tcdc.level == "entire atmosphere"
end
test "includes precipitation at surface" do
messages = HrrrClient.surface_messages()
apcp = Enum.find(messages, &(&1.var == "APCP"))
assert apcp.level == "surface"
end
end
describe "download_grib_ranges_to_file/3" do
test "writes bytes to file in offset-ascending order even when fetched out of order" do
# Two non-adjacent ranges (so they don't get merged). Responses
# deliberately return with the second range "arriving" first to
# simulate out-of-order completion under parallel fetches; the
# final file must still be ordered by offset.
ranges = [{0, 9}, {100, 119}]
range_a = String.duplicate("A", 10)
range_b = String.duplicate("B", 20)
Req.Test.stub(HrrrClient, fn conn ->
[range] = Plug.Conn.get_req_header(conn, "range")
case range do
"bytes=0-9" ->
Plug.Conn.send_resp(conn, 206, range_a)
"bytes=100-119" ->
Plug.Conn.send_resp(conn, 206, range_b)
end
end)
tmp = Path.join(System.tmp_dir!(), "hrrr_download_test_#{System.unique_integer([:positive])}.grib2")
try do
assert :ok = HrrrClient.download_grib_ranges_to_file("http://test/grib", ranges, tmp)
assert File.read!(tmp) == range_a <> range_b
after
File.rm(tmp)
end
end
end
describe "merge_grid_data/2" do
test "merges surface and pressure grids by point key" do
point_a = {32.90, -97.04}
point_b = {33.10, -96.80}
sfc_grid = %{
point_a => %{
"TMP:2 m above ground" => 299.0,
"DPT:2 m above ground" => 292.0,
"PRES:surface" => 101_350.0,
"HPBL:surface" => 1500.0,
"PWAT:entire atmosphere (considered as a single layer)" => 25.0,
"UGRD:10 m above ground" => 3.5,
"VGRD:10 m above ground" => -2.1,
"TCDC:entire atmosphere" => 75.0,
"APCP:surface" => 1.2
},
point_b => %{
"TMP:2 m above ground" => 297.0,
"DPT:2 m above ground" => 290.0,
"PRES:surface" => 101_200.0,
"HPBL:surface" => 1200.0,
"PWAT:entire atmosphere (considered as a single layer)" => 22.0,
"UGRD:10 m above ground" => 2.0,
"VGRD:10 m above ground" => -1.5,
"TCDC:entire atmosphere" => 50.0,
"APCP:surface" => 0.0
}
}
prs_grid = %{
point_a => %{
"TMP:1000 mb" => 298.0,
"DPT:1000 mb" => 291.0,
"HGT:1000 mb" => 110.0
},
point_b => %{
"TMP:1000 mb" => 296.0,
"DPT:1000 mb" => 289.0,
"HGT:1000 mb" => 105.0
}
}
merged = HrrrClient.merge_grid_data(sfc_grid, prs_grid)
assert map_size(merged) == 2
assert merged[point_a]["TMP:2 m above ground"] == 299.0
assert merged[point_a]["TMP:1000 mb"] == 298.0
assert merged[point_b]["TMP:2 m above ground"] == 297.0
assert merged[point_b]["TMP:1000 mb"] == 296.0
end
test "handles empty pressure data" do
point = {32.90, -97.04}
sfc_grid = %{
point => %{"TMP:2 m above ground" => 299.0}
}
merged = HrrrClient.merge_grid_data(sfc_grid, %{})
assert map_size(merged) == 1
assert merged[point]["TMP:2 m above ground"] == 299.0
end
end
end