diff --git a/config/test.exs b/config/test.exs index 7bcd4612..6243114e 100644 --- a/config/test.exs +++ b/config/test.exs @@ -65,6 +65,7 @@ config :microwaveprop, narr_req_options: [plug: {Req.Test, Microwaveprop.Weather # Route HTTP requests through Req.Test stubs 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, srtm_req_options: [plug: {Req.Test, Microwaveprop.Terrain.Srtm}, retry: false] config :microwaveprop, start_freshness_monitor: false diff --git a/lib/microwaveprop/weather/rtma_client.ex b/lib/microwaveprop/weather/rtma_client.ex index 79797e5c..6ebb3fd3 100644 --- a/lib/microwaveprop/weather/rtma_client.ex +++ b/lib/microwaveprop/weather/rtma_client.ex @@ -72,7 +72,7 @@ defmodule Microwaveprop.Weather.RtmaClient do end 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: status}} -> {:error, "RTMA idx HTTP #{status}"} {:error, reason} -> {:error, "RTMA idx failed: #{inspect(reason)}"} @@ -118,9 +118,12 @@ defmodule Microwaveprop.Weather.RtmaClient do ranges |> Task.async_stream( fn {range_start, range_end} -> - Req.get(url, - headers: [{"Range", "bytes=#{range_start}-#{range_end}"}], - receive_timeout: 30_000 + Req.get( + url, + [ + headers: [{"Range", "bytes=#{range_start}-#{range_end}"}], + receive_timeout: 30_000 + ] ++ req_options() ) end, max_concurrency: 4, @@ -159,4 +162,6 @@ defmodule Microwaveprop.Weather.RtmaClient do defp pa_to_mb(nil), do: nil defp pa_to_mb(pa), do: pa / 100.0 + + defp req_options, do: Application.get_env(:microwaveprop, :rtma_req_options, []) end diff --git a/test/microwaveprop/weather/rtma_client_test.exs b/test/microwaveprop/weather/rtma_client_test.exs new file mode 100644 index 00000000..d8dcae43 --- /dev/null +++ b/test/microwaveprop/weather/rtma_client_test.exs @@ -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 diff --git a/test/microwaveprop/workers/rtma_fetch_worker_test.exs b/test/microwaveprop/workers/rtma_fetch_worker_test.exs new file mode 100644 index 00000000..eeb498f2 --- /dev/null +++ b/test/microwaveprop/workers/rtma_fetch_worker_test.exs @@ -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