test: cover rtma_client + rtma_fetch_worker, add Req.Test seam
RtmaClient was the only weather client without a Req.Test plug seam — added the standard req_options() helper matching NarrClient/HrrrClient/ etc. so its HTTP calls can be stubbed in test. 17 characterization tests: URL construction, idx fetch 404/503, hour truncation, malformed idx, range-GET error propagation, worker's existing-row short-circuit, client-error propagation. GRIB2 byte parsing paths deferred to the wgrib2 coverage task.
This commit is contained in:
parent
923575c392
commit
cb82160447
4 changed files with 331 additions and 4 deletions
|
|
@ -65,6 +65,7 @@ config :microwaveprop, narr_req_options: [plug: {Req.Test, Microwaveprop.Weather
|
||||||
|
|
||||||
# Route HTTP requests through Req.Test stubs
|
# Route HTTP requests through Req.Test stubs
|
||||||
config :microwaveprop, nexrad_req_options: [plug: {Req.Test, Microwaveprop.Weather.NexradClient}, retry: false]
|
config :microwaveprop, nexrad_req_options: [plug: {Req.Test, Microwaveprop.Weather.NexradClient}, retry: false]
|
||||||
|
config :microwaveprop, rtma_req_options: [plug: {Req.Test, Microwaveprop.Weather.RtmaClient}, retry: false]
|
||||||
config :microwaveprop, solar_req_options: [plug: {Req.Test, Microwaveprop.Weather.SolarClient}]
|
config :microwaveprop, solar_req_options: [plug: {Req.Test, Microwaveprop.Weather.SolarClient}]
|
||||||
config :microwaveprop, srtm_req_options: [plug: {Req.Test, Microwaveprop.Terrain.Srtm}, retry: false]
|
config :microwaveprop, srtm_req_options: [plug: {Req.Test, Microwaveprop.Terrain.Srtm}, retry: false]
|
||||||
config :microwaveprop, start_freshness_monitor: false
|
config :microwaveprop, start_freshness_monitor: false
|
||||||
|
|
|
||||||
|
|
@ -72,7 +72,7 @@ defmodule Microwaveprop.Weather.RtmaClient do
|
||||||
end
|
end
|
||||||
|
|
||||||
defp fetch_idx(url) do
|
defp fetch_idx(url) do
|
||||||
case Req.get(url, receive_timeout: 15_000) do
|
case Req.get(url, [receive_timeout: 15_000] ++ req_options()) do
|
||||||
{:ok, %{status: 200, body: body}} -> {:ok, body}
|
{:ok, %{status: 200, body: body}} -> {:ok, body}
|
||||||
{:ok, %{status: status}} -> {:error, "RTMA idx HTTP #{status}"}
|
{:ok, %{status: status}} -> {:error, "RTMA idx HTTP #{status}"}
|
||||||
{:error, reason} -> {:error, "RTMA idx failed: #{inspect(reason)}"}
|
{:error, reason} -> {:error, "RTMA idx failed: #{inspect(reason)}"}
|
||||||
|
|
@ -118,9 +118,12 @@ defmodule Microwaveprop.Weather.RtmaClient do
|
||||||
ranges
|
ranges
|
||||||
|> Task.async_stream(
|
|> Task.async_stream(
|
||||||
fn {range_start, range_end} ->
|
fn {range_start, range_end} ->
|
||||||
Req.get(url,
|
Req.get(
|
||||||
headers: [{"Range", "bytes=#{range_start}-#{range_end}"}],
|
url,
|
||||||
receive_timeout: 30_000
|
[
|
||||||
|
headers: [{"Range", "bytes=#{range_start}-#{range_end}"}],
|
||||||
|
receive_timeout: 30_000
|
||||||
|
] ++ req_options()
|
||||||
)
|
)
|
||||||
end,
|
end,
|
||||||
max_concurrency: 4,
|
max_concurrency: 4,
|
||||||
|
|
@ -159,4 +162,6 @@ defmodule Microwaveprop.Weather.RtmaClient do
|
||||||
|
|
||||||
defp pa_to_mb(nil), do: nil
|
defp pa_to_mb(nil), do: nil
|
||||||
defp pa_to_mb(pa), do: pa / 100.0
|
defp pa_to_mb(pa), do: pa / 100.0
|
||||||
|
|
||||||
|
defp req_options, do: Application.get_env(:microwaveprop, :rtma_req_options, [])
|
||||||
end
|
end
|
||||||
|
|
|
||||||
214
test/microwaveprop/weather/rtma_client_test.exs
Normal file
214
test/microwaveprop/weather/rtma_client_test.exs
Normal file
|
|
@ -0,0 +1,214 @@
|
||||||
|
defmodule Microwaveprop.Weather.RtmaClientTest do
|
||||||
|
use ExUnit.Case, async: true
|
||||||
|
|
||||||
|
alias Microwaveprop.Weather.RtmaClient
|
||||||
|
|
||||||
|
describe "rtma_url/1" do
|
||||||
|
test "builds the S3 URL for an RTMA analysis time" do
|
||||||
|
vt = ~U[2026-04-15 18:00:00Z]
|
||||||
|
|
||||||
|
assert RtmaClient.rtma_url(vt) ==
|
||||||
|
"https://noaa-rtma-pds.s3.amazonaws.com/rtma2p5.20260415/rtma2p5.t18z.2dvaranl_ndfd.grb2_wexp"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "zero-pads single-digit hours" do
|
||||||
|
vt = ~U[2026-04-15 03:00:00Z]
|
||||||
|
|
||||||
|
assert RtmaClient.rtma_url(vt) ==
|
||||||
|
"https://noaa-rtma-pds.s3.amazonaws.com/rtma2p5.20260415/rtma2p5.t03z.2dvaranl_ndfd.grb2_wexp"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "handles hour zero" do
|
||||||
|
vt = ~U[2026-04-15 00:00:00Z]
|
||||||
|
|
||||||
|
assert RtmaClient.rtma_url(vt) ==
|
||||||
|
"https://noaa-rtma-pds.s3.amazonaws.com/rtma2p5.20260415/rtma2p5.t00z.2dvaranl_ndfd.grb2_wexp"
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "fetch_observation/3 — idx layer" do
|
||||||
|
test "returns {:error, _} when the idx fetch returns 404 (not yet published)" do
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 404, "not found")
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 500, "idx should have short-circuited")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, reason} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
|
||||||
|
assert reason =~ "RTMA idx HTTP 404"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns {:error, _} when the idx fetch returns a non-200 status" do
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 503, "unavailable")
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, reason} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
|
||||||
|
assert reason =~ "RTMA idx HTTP 503"
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "fetch_observation/3 — hour truncation" do
|
||||||
|
test "truncates minutes and seconds to the analysis hour in the request URL" do
|
||||||
|
test_pid = self()
|
||||||
|
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
send(test_pid, {:request_path, conn.request_path, conn.method})
|
||||||
|
# Return an empty idx so downstream stages short-circuit on 0 ranges.
|
||||||
|
Plug.Conn.send_resp(conn, 200, "")
|
||||||
|
end)
|
||||||
|
|
||||||
|
# 18:47:33 must request the 18z (not 19z) analysis file — truncation
|
||||||
|
# is load-bearing for matching the stored row's hour-aligned valid_time.
|
||||||
|
assert {:error, _} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:47:33Z])
|
||||||
|
|
||||||
|
assert_received {:request_path, path, "GET"}
|
||||||
|
assert String.contains?(path, "rtma2p5.t18z")
|
||||||
|
refute String.contains?(path, "rtma2p5.t19z")
|
||||||
|
end
|
||||||
|
|
||||||
|
test "already-aligned hour passes through unchanged" do
|
||||||
|
test_pid = self()
|
||||||
|
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
send(test_pid, {:request_path, conn.request_path})
|
||||||
|
Plug.Conn.send_resp(conn, 200, "")
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, _} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 12:00:00Z])
|
||||||
|
|
||||||
|
assert_received {:request_path, path}
|
||||||
|
assert String.contains?(path, "rtma2p5.t12z")
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "fetch_observation/3 — extract layer" do
|
||||||
|
# With an empty (or non-matching) idx body, `byte_ranges_for_fields/2`
|
||||||
|
# produces zero ranges, so `download_ranges/2` does no HTTP and returns
|
||||||
|
# {:ok, ""}, which the GRIB2 Extractor splits into an empty message
|
||||||
|
# list → {:ok, %{}} → {:error, "RTMA: no data..."} path.
|
||||||
|
test "returns {:error, 'RTMA: no data ...'} when the extractor yields no fields" do
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
# idx contains no wanted fields — produces zero byte-ranges.
|
||||||
|
Plug.Conn.send_resp(conn, 200, "1:0:d=2026041518:FOOBAR:nothing:anl:\n")
|
||||||
|
else
|
||||||
|
# Should not be reached because no ranges means no GET.
|
||||||
|
Plug.Conn.send_resp(conn, 500, "unexpected GRB request")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, reason} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
|
||||||
|
assert reason =~ "RTMA: no data"
|
||||||
|
# rlat/rlon are reported to 2.5 km precision (1/40°).
|
||||||
|
assert reason =~ "32.9"
|
||||||
|
assert reason =~ "-97.0"
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns {:error, 'RTMA: no data ...'} when grib bytes are non-GRIB garbage" do
|
||||||
|
# Serve an idx that DOES match wanted fields so byte-ranges are non-empty,
|
||||||
|
# then serve junk bytes on the ranged GET. The GRIB2 Extractor's
|
||||||
|
# split_messages/1 skips unknown bytes and returns an empty message list,
|
||||||
|
# so extract_point still yields the "no data" error.
|
||||||
|
idx = """
|
||||||
|
1:0:d=2026041518:TMP:2 m above ground:anl:
|
||||||
|
2:100:d=2026041518:DPT:2 m above ground:anl:
|
||||||
|
3:200:d=2026041518:PRES:surface:anl:
|
||||||
|
"""
|
||||||
|
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 200, idx)
|
||||||
|
else
|
||||||
|
# 206 Partial Content with non-GRIB junk.
|
||||||
|
Plug.Conn.send_resp(conn, 206, :binary.copy(<<0>>, 1024))
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, reason} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
|
||||||
|
assert reason =~ "RTMA: no data"
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "fetch_observation/3 — malformed idx" do
|
||||||
|
test "crashes on a malformed idx line with a non-integer offset" do
|
||||||
|
# byte_ranges_for_fields/2 calls String.to_integer/1 on the offset
|
||||||
|
# field without a guard — a non-numeric offset is a programmer/server
|
||||||
|
# contract violation and the process raises. Characterize this so
|
||||||
|
# any future change to forgiving-parsing is intentional.
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 200, "1:notanumber:d=2026041518:TMP:2 m above ground:anl:\n")
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert_raise ArgumentError, fn ->
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
test "tolerates an idx line with missing trailing fields (uses empty field name)" do
|
||||||
|
# A single-field line falls back to empty strings for var/level via
|
||||||
|
# Enum.at/3 defaults, so it simply doesn't match any wanted field.
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 200, "1:0\n")
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, reason} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
|
||||||
|
# Empty idx body → zero ranges → empty GRIB → "no data" at extraction.
|
||||||
|
assert reason =~ "RTMA: no data"
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
describe "fetch_observation/3 — download layer" do
|
||||||
|
# download_ranges/2's Task.async_stream reduces over results; any
|
||||||
|
# non-200/206 downgrades the accumulator to {:error, _}. But a catch-all
|
||||||
|
# `_, acc -> acc` clause in the reducer silently swallows :exit/:timeout
|
||||||
|
# task results. If all tasks time out, the accumulator stays {:ok, []},
|
||||||
|
# which produces an empty binary and ultimately "RTMA: no data ...".
|
||||||
|
# Characterizing, not prescribing — tighten in a follow-up if desired.
|
||||||
|
test "surfaces a non-200 status from the ranged GET as an error" do
|
||||||
|
idx = """
|
||||||
|
1:0:d=2026041518:TMP:2 m above ground:anl:
|
||||||
|
2:100:d=2026041518:DPT:2 m above ground:anl:
|
||||||
|
"""
|
||||||
|
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 200, idx)
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 500, "s3 blew up")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
assert {:error, reason} =
|
||||||
|
RtmaClient.fetch_observation(32.9, -97.0, ~U[2026-04-15 18:00:00Z])
|
||||||
|
|
||||||
|
assert reason =~ "RTMA download HTTP 500"
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
107
test/microwaveprop/workers/rtma_fetch_worker_test.exs
Normal file
107
test/microwaveprop/workers/rtma_fetch_worker_test.exs
Normal file
|
|
@ -0,0 +1,107 @@
|
||||||
|
defmodule Microwaveprop.Workers.RtmaFetchWorkerTest do
|
||||||
|
use Microwaveprop.DataCase, async: false
|
||||||
|
use Oban.Testing, repo: Microwaveprop.Repo
|
||||||
|
|
||||||
|
alias Microwaveprop.Weather.RtmaClient
|
||||||
|
alias Microwaveprop.Weather.RtmaObservation
|
||||||
|
alias Microwaveprop.Workers.RtmaFetchWorker
|
||||||
|
|
||||||
|
describe "perform/1" do
|
||||||
|
test "skips the fetch (no HTTP) when a matching row already exists in the snapped bounding box" do
|
||||||
|
valid_time = ~U[2026-04-15 18:00:00Z]
|
||||||
|
|
||||||
|
# Pre-populate a row at the snapped coordinates (32.9 → 32.9 at 1/40°,
|
||||||
|
# -97.0 → -97.0). The worker's bounding-box short-circuit must find it
|
||||||
|
# and return :ok without contacting RtmaClient.
|
||||||
|
%RtmaObservation{}
|
||||||
|
|> RtmaObservation.changeset(%{
|
||||||
|
lat: 32.9,
|
||||||
|
lon: -97.0,
|
||||||
|
valid_time: valid_time,
|
||||||
|
temp_c: 20.0
|
||||||
|
})
|
||||||
|
|> Repo.insert!()
|
||||||
|
|
||||||
|
Req.Test.stub(RtmaClient, fn _conn ->
|
||||||
|
raise "RtmaClient should not be contacted when an observation already exists"
|
||||||
|
end)
|
||||||
|
|
||||||
|
args = %{
|
||||||
|
"lat" => 32.9,
|
||||||
|
"lon" => -97.0,
|
||||||
|
"valid_time" => DateTime.to_iso8601(valid_time)
|
||||||
|
}
|
||||||
|
|
||||||
|
assert :ok = RtmaFetchWorker.perform(%Oban.Job{args: args})
|
||||||
|
assert Repo.aggregate(RtmaObservation, :count, :id) == 1
|
||||||
|
end
|
||||||
|
|
||||||
|
# With garbage-bytes-on-the-wire (no real GRIB2 fixture), the extractor
|
||||||
|
# yields no fields and RtmaClient returns {:error, "RTMA: no data ..."},
|
||||||
|
# which the worker surfaces as {:error, _}. This characterizes that the
|
||||||
|
# worker propagates the client error (and does NOT insert a blank row).
|
||||||
|
test "returns {:error, _} and inserts nothing when the client fails to extract" do
|
||||||
|
idx = """
|
||||||
|
1:0:d=2026041518:TMP:2 m above ground:anl:
|
||||||
|
2:100:d=2026041518:DPT:2 m above ground:anl:
|
||||||
|
"""
|
||||||
|
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 200, idx)
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 206, :binary.copy(<<0>>, 1024))
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
args = %{
|
||||||
|
"lat" => 32.9,
|
||||||
|
"lon" => -97.0,
|
||||||
|
"valid_time" => "2026-04-15T18:00:00Z"
|
||||||
|
}
|
||||||
|
|
||||||
|
assert {:error, reason} = RtmaFetchWorker.perform(%Oban.Job{args: args})
|
||||||
|
assert reason =~ "RTMA: no data"
|
||||||
|
assert Repo.aggregate(RtmaObservation, :count, :id) == 0
|
||||||
|
end
|
||||||
|
|
||||||
|
test "returns {:error, _} when upstream idx returns 404" do
|
||||||
|
Req.Test.stub(RtmaClient, fn conn ->
|
||||||
|
if String.ends_with?(conn.request_path, ".idx") do
|
||||||
|
Plug.Conn.send_resp(conn, 404, "not yet published")
|
||||||
|
else
|
||||||
|
Plug.Conn.send_resp(conn, 500, "unexpected")
|
||||||
|
end
|
||||||
|
end)
|
||||||
|
|
||||||
|
args = %{
|
||||||
|
"lat" => 32.9,
|
||||||
|
"lon" => -97.0,
|
||||||
|
"valid_time" => "2026-04-15T18:00:00Z"
|
||||||
|
}
|
||||||
|
|
||||||
|
assert {:error, reason} = RtmaFetchWorker.perform(%Oban.Job{args: args})
|
||||||
|
assert reason =~ "RTMA idx HTTP 404"
|
||||||
|
assert Repo.aggregate(RtmaObservation, :count, :id) == 0
|
||||||
|
end
|
||||||
|
|
||||||
|
test "raises FunctionClauseError when args are missing required keys" do
|
||||||
|
# The worker only matches the shape with lat/lon/valid_time — missing
|
||||||
|
# keys must crash loudly rather than silently succeed. This pins the
|
||||||
|
# current fail-fast contract.
|
||||||
|
assert_raise FunctionClauseError, fn ->
|
||||||
|
RtmaFetchWorker.perform(%Oban.Job{args: %{}})
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
|
test "raises when valid_time is not an ISO8601 string" do
|
||||||
|
# DateTime.from_iso8601/1 returns {:error, _}, and the `{:ok, _, _} =`
|
||||||
|
# match raises — another fail-fast contract worth pinning.
|
||||||
|
assert_raise MatchError, fn ->
|
||||||
|
RtmaFetchWorker.perform(%Oban.Job{
|
||||||
|
args: %{"lat" => 32.9, "lon" => -97.0, "valid_time" => "not-a-timestamp"}
|
||||||
|
})
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
|
end
|
||||||
Loading…
Add table
Reference in a new issue